消息队列 Kafka版
消息队列 Kafka版
消息队列 Kafka版
文档指南
请输入
用户指南
实例管理
Kafka Agent
API 参考
V2（2022-05-01）
OpenAPIV2
网络管理
Topic 管理
Group 管理
SASL 用户管理
白名单管理
附录
V1（2018-01-01）
实例管理
Topic 管理
Group 管理
SDK 参考
Apache Kafka SDK
Java 客户端 SDK
Go 客户端 SDK
Python 客户端 SDK
Volcengine SDK
Volcengine Python SDK
Volcengine Java SDK
最佳实践
- 文档首页
消息队列 Kafka版SDK 参考Apache Kafka SDKPython 客户端 SDKSASL_SSL 接入点 PLAIN 机制收发消息
复制全文
下载 pdfSASL_SSL 接入点 PLAIN 机制收发消息
文档反馈
问问助手本文以 Python 客户端为例，介绍如何在 VPC 或公网环境下通过 SASL_SSL 接入点 PLAIN 机制接入消息队列 Kafka版，并收发消息。
前提条件
1 添加配置文件
{
"bootstrap.servers": "xxxxx", // 修改配置为实例的SASL接入点
"security.protocol": "SASL_SSL", // 固定为SASL_SSL
"ssl.endpoint.identification.algorithm": "none", //跳过SSL证书域名校验
"topic": "xxxx", // 修改配置为待发送的topic名称
"consumer": {
"group.id": "xxxx" // 修改为指定消费组的名称
},
"sasl": {
"mechanism": "PLAIN", // 用户类型为Plain时固定为PLAIN
"username": "xxxx", // PLAIN用户名
"password": "xxxx" // PLAIN用户密码
}
}
2 发送消息
实现方法
- 创建消息发送程序 producer.py。
- 编译并运行 producer.py 发送消息。
- 查看运行结果。
- 运行结果示例如下。
-
- 消息队列 Kafka版提供示例项目供您快速接入，下载并解压缩 Demo 后，可以直接执行以下命令发送并消费消息。
- python3 {DemoPath}/bytedance_kafka.py
示例代码
通过 SASL_SSL 接入点生产消息的示例代码如下，您也可以参考 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 消费消息。
- 查看运行结果。
示例代码
通过 SASL_SSL 接入点消费消息的示例代码如下，您也可以参考 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()
最近更新时间：2026.04.29 15:41:35
这个页面对您有帮助吗？
有用
有用
无用
无用