Message Queue for Kafka
Message Queue for Kafka
Document
- Documentation
Message Queue for KafkaSDK ReferenceApache Kafka SDKPython client SDKSASL_SSL access point (SCRAM mechanism) message production and consumption
Copy page
Download PDFPython client SDK
SASL_SSL access point (SCRAM mechanism) message production and consumption
Copy pageDownload PDFSASL_SSL access point (SCRAM mechanism) message production and consumption
This topic uses the Python client as an example to describe how to connect to Message Queue for Kafka using the SCRAM mechanism through SASL_SSL endpoints in VPC or public network environments, and send and receive messages.
Prerequisites
1. Add a configuration file
Create the Message Queue for Kafka configuration file config.json. When connecting to a SASL access point using the SCRAM mechanism, the following is an example configuration file. For detailed explanations of the configuration file fields, see SDK configuration description.
tip
- Fill in the related parameters according to the comments and remove the comments.
- Under SCRAM mechanism, use a SCRAM user with corresponding Topic access rights for SASL authentication. For how to obtain the username and password, refer to 2 Collect Connection Information.
{
"bootstrap.servers": "xxxxx",
"security.protocol": "SASL_SSL",
"ssl.endpoint.identification.algorithm": "none",
"topic": "xxxx",
"consumer": {
"group.id": "xxxx"
},
"sasl": {
"enabled": true,
"mechanism": "SCRAM-SHA-256",
"username": "xxxx",
"password": "xxxx"
}
}
2 Send messages
Implementation method
- Create the message sending program producer.py.
- Compile and run producer.py to send messages.
- Check the running results.
- The running result example is as follows.
-
tip
Message queue for Kafka provides an example project for quick integration. After downloading and extracting the demo, you can directly execute the following commands to send and consume messages.
python3 {DemoPath}/bytedance_kafka.py
Example code
The following example code demonstrates how to produce messages using the SCRAM mechanism via a SASL_SSL access point. You can also refer to the example file {DemoPath}/client/producer.py to implement the relevant business logic.
from confluent_kafka import Producer
def callback(err, meta):
"""
py:function:: callback(err, meta)
Handle the result of message delivery.
:param confluent_kafka.KafkaError err: error if delivery is failed
:param confluent_kafka.Message meta: message metadata if delivery is success
:return: None
"""
if err is None:
print('[INFO] Delivered message to topic {} [{}] at offset {}'
.format(meta.topic(), meta.partition(), meta.offset()))
else:
print('[ERROR] Delivery failed: {}'.format(err))
def run_producer(conf):
# create producer instance
p = Producer(conf.producer_conf())
for i in range(10):
# send a message
p.produce(conf.topic(), 'demon message {}'.format(i).encode('utf-8'), on_delivery=callback)
# call poll method to handle deliver result
p.poll(timeout=0)
# flush all messages
p.flush(timeout=5.0)
3 consume messages
Implementation method
- Create a Consumer subscription message program consumer.py.
- Compile and run consumer.py to consume messages.
- Check the running results.
Example code
The following example code demonstrates how to consume messages using the SCRAM mechanism via a SASL_SSL endpoint. You can also refer to the example file {DemoPath}/client/consumer.py to implement the relevant business logic.
from confluent_kafka import Consumer
def handle_message(msg):
"""
py:function:: handle_message(msg)
Handle consumed message.
:param confluent_kafka.Message msg: consumed message meta.
:return: None
"""
print('[INFO] Consumed a message from topic {} [{}] at offset {}. value: {}'.
format(msg.topic(), msg.partition(), msg.offset(), msg.value().decode('utf-8')))
def run_consumer(conf):
# create a consumer instance
c = Consumer(conf.consumer_conf())
# subscribe topic
c.subscribe([conf.topic()])
try:
count = 0
while count < 10:
msg = c.poll(timeout=1.0)
if msg is None:
# consumer initial and group rebalance may take some time.
continue
if msg.error():
print('[ERROR] Consume msg failed, {}'.format(msg.error()))
continue
# handle kafka messages. this
handle_message(msg)
count += 1
c.commit(asynchronous=False)
finally:
c.close()
Last updated: 2026.04.29 18:34:22
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
