Python操作Kafka例子

最新推荐文章于 2025-04-03 18:14:56 发布

伍迪艾伦学Python

最新推荐文章于 2025-04-03 18:14:56 发布

阅读量3.2k

点赞数

文章标签： python

本文链接：https://blog.csdn.net/pppython123/article/details/123643946

版权

Kafka作为消息中间件是非常受欢迎的，对应Python库名为kafka-python；

import json
import traceback
from kafka.errors import kafka_errors
from kafka import KafkaProducer, KafkaConsumer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'], 
                         key_serializer=lambda key: json.dumps(key).encode(),
                         value_serializer=lambda value: json.dumps(value).encode())

for i in range(3):
    msg = 'message ' + str(i)
    future = producer.send('kafka_topic', key='count', value=msg)  
    print(f"send {msg}")
    try:
        future.get(timeout=10)    # 看是否发送成功           
    except kafka_errors:
        traceback.print_exc()

上段为Kafka生产者，下段对应Kafka消费者，KafkaConsumer会进入持续接收状态；

consumer = KafkaConsumer('kafka_topic', bootstrap_servers=['localhost:9092'], group_id='test')
for msg in consumer:
    print(f"receive, key: {json.loads(msg.key.decode())}, value: {json.loads(msg.value.decode())}")