测开之路七十四:python处理kafka


 

kafka-python地址:https://github.com/dpkp/kafka-python

安装kafka-python:pip install kafka-python

 

接收消息

from kafka import KafkaConsumer
consumer = KafkaConsumer('test', bootstrap_servers=['localhost:9092']) # 定义消费者
for msg in consumer:
print(msg)

发条消息尝试接收,结果是byte类型

 

发送消息

from kafka import KafkaProducer
import json

data = '123qweasd45'
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8'))
# value_serializer=lambda v: json.dumps(v).encode('utf-8'):由于默认发送的数据是byte,这里把要发送的数据序列化成json
producer.send('test', json.dumps(data), partition=0)
producer.close()

执行投递消息,查看消费者的输出

窗口也可以展示

 


注意!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系我们删除。



 
  © 2014-2022 ITdaan.com 联系我们: