消息队列 Kafka版
消息队列 Kafka版
文档指南
请输入
- 文档首页
消息队列 Kafka版SDK 参考Apache Kafka SDKPython 客户端 SDKPLAINTEXT 接入点收发消息
复制全文
下载 pdfPLAINTEXT 接入点收发消息
文档反馈
问问助手本文以 Python 客户端为例，介绍如何在 VPC 环境下通过 PLAINTEXT 接入点接入消息队列 Kafka版，并收发消息。
前提条件
1 添加配置文件
使用 PLAINTEXT 接入点时，配置文件示例如下。
说明
请根据注释提示填写相关参数，并删除注释。
{
"bootstrap.servers": "xxxxx", // 修改配置为实例的 PLAINTEXT 接入点
"security.protocol": "PLAINTEXT", // PLAINTEXT 接入点访问时，固定设置为 PLAINTEXT
"topic": "xxxx", // 修改配置为待发送的topic名称
"consumer": {
"group.id": "xxxx" // 修改为指定消费组的名称
}
}
2 发送消息
实现方法
- 创建消息发送程序 producer.py。
- 编译并运行 producer.py 发送消息。
- 查看运行结果。
- 运行结果示例如下。
-
说明
python3 {DemoPath}/bytedance_kafka.py
示例代码
通过 PLAINTEXT 接入点生产消息的示例代码如下，您也可以参考 Demo 中的示例文件 {DemoPath}/client/producer.py，实现相关业务逻辑。
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 消费消息
实现方法
- 创建 Consumer 订阅消息程序 consumer.py。
- 编译并运行 consumer.py 消费消息。
- 查看运行结果。
消息消费示例代码
通过 PLAINTEXT 接入点消费消息的示例代码如下，您也可以参考 Demo 中的示例文件 {DemoPath}/client/consumer.py，实现相关业务逻辑。
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()
最近更新时间：2025.08.08 17:47:27
这个页面对您有帮助吗？
有用
有用
无用
无用