使用python操作kafka

标签：import   操作   pytho   安装   val   hold   消息   consumer   没有   

使用python操作kafka目前比较常用的库是kafka-python库

安装kafka-python

pip3 install kafka-python

生产者

producer_test.py

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='192.168.0.121:9092')  # 连接kafka

msg = "Hello World".encode('utf-8')  # 发送内容,必须是bytes类型
producer.send('test', msg)  # 发送的topic为test
producer.close()

执行此程序，它没有输出！这个是正常的

消费者

from kafka import KafkaConsumer

consumer = KafkaConsumer('test', bootstrap_servers=['192.168.0.121:9092'])
for msg in consumer:
    recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)
    print(recv)

执行此程序，此时会hold住，因为它在等待生产者发送消息！

再次执行生产者，此时会输出：

test:0:9: key=None value=b'Hello World'

使用python操作kafka

标签：import   操作   pytho   安装   val   hold   消息   consumer   没有   

原文地址：https://www.cnblogs.com/sanduzxcvbnm/p/11579199.html