Message Queue for Kafka
Message Queue for Kafka
Document
- Documentation
Message Queue for KafkaSDK ReferenceApache Kafka SDKPython client SDKSASL_SSL access point (PLAIN mechanism) message production and consumption
Copy page
Download PDFPython client SDK
SASL_SSL access point (PLAIN mechanism) message production and consumption
Copy pageDownload PDFSASL_SSL access point (PLAIN 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 SASL_SSL access point and the PLAIN mechanism 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. An example configuration file is as follows when connecting using the PLAIN mechanism through a SASL_SSL endpoint. 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 the PLAIN mechanism, SASL authentication should be performed using PLAIN users with the appropriate Topic access permissions. 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": {
"mechanism": "PLAIN",
"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
Sample code
The following sample code demonstrates how to produce messages through a SASL_SSL endpoint. You can also refer to the sample 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.
Sample code
The sample code for consuming messages through a SASL_SSL endpoint is as follows. You can also refer to the sample file in the demo, {DemoPath}/client/consumer.py, to implement related 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
