Python 连接 Kafka 发送接收消息数据

Kafka模块

pip install kafka==1.3.5
pip install kafka-python==1.3.5

生产者

#!/usr/bin/env python
#encoding:utf8
#author: zeping lai
#生产者

from kafka import KafkaProducer 
import random

kafka_topic = "test-topic-zeze"
kafka_bootstrap_servers = ['192.168.100.21:9092','192.168.100.22:9092','192.168.100.23:9092']

producer = KafkaProducer(bootstrap_servers=kafka_bootstrap_servers)
phones = [ "huawei", "xiaomi", "apple" ]
for p in phones:
     p = p + "-" + str(random.randint(1,5))
     msg = bytes(p, encoding='utf-8')
     print(msg)
     producer.send(kafka_topic, key=b"iphone", value=msg)
producer.close()

消费者

#!/usr/bin/env python
#encoding:utf8
#author: zeping lai

kafka_topic = "test-topic-zeze"
kafka_topic_group = "test-group-zeze"
kafka_bootstrap_servers = ['192.168.100.21:9092','192.168.100.22:9092','192.168.100.23:9092']

from kafka import KafkaConsumer
consumer = KafkaConsumer(kafka_topic,group_id=kafka_topic_group,bootstrap_servers=kafka_bootstrap_servers)

for message in consumer:
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))