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

Posted 2022-06-03 zhongyehai

 

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

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

窗口也可以展示