Python使用kafka-python库发送和接收JSON数据的完整示例
本文介绍如何使用Python的kafka-python库实现Kafka消息的JSON格式读写。重点解析生产者KafkaProducer中value_serializer参数的配置,通过json.dumps将字典编码为UTF-8字节流;以及消费者KafkaConsumer中value_deserializer参数的设置,利用json.loads解析消息。示例包含完整的连接配置、数据构造及循环发送逻辑,适用于需要结构化数据传输的场景。
具体实现代码如下所示:
安装kafka支持库pip install kafka-python
from kafka import KafkaProducer
import json
'''
生产者demo
向test_lyl2主题中循环写入10条json数据
注意事项:要写入json数据需加上value_serializer参数,如下代码
'''
producer = KafkaProducer(
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
bootstrap_servers=['192.168.12.101:6667','192.168.12.102:6667','192.168.12.103:6667']
)
for i in range(10):
data={
"name":"李四",
"age":23,
"gender":"男",
"id":i
}
producer.send('test_lyl2', data)
producer.close()
from kafka import KafkaConsumer
import json
'''
消费者demo
消费test_lyl2主题中的数据
注意事项:如需以json格式读取数据需加上value_deserializer参数
'''
consumer = KafkaConsumer('test_lyl2',group_id="lyl-gid1",
bootstrap_servers=['192.168.12.101:6667','192.168.12.102:6667','192.168.12.103:6667'],
auto_offset_reset='earliest',value_deserializer=json.loads
)
for message in consumer:
print(message.value)
上述内容即为Python操作Kafka写入JSON数据的简易演示。希望该示例能为读者提供有价值的参考,欢迎持续关注以获取更多技术分享。
免责声明:本文内容仅供参考,不代表本站观点。转载或引用请注明出处。









