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

时间:2022-12-10 13:08:43

 

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

安装kafka-python:pip install kafka-python

 

接收消息

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

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

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

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

 

发送消息

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

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()

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

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

窗口也可以展示

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