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

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

安装kafka-python：pip install kafka-python

接收消息

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

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

发送消息

from kafka import KafkaProducerimport 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，这里把要发送的数据序列化成jsonproducer.send(‘test‘, json.dumps(data), partition=0)producer.close()

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

窗口也可以展示

原文地址：https://www.cnblogs.com/zhongyehai/p/11285963.html