消费者:
from kafka import KafkaConsumer
consumer=KafkaConsumer(bootstrap_servers= ['192.168.0.0','192.168.0.1','192.168.0.2'],group_id='test',auto_offset_reset='earliest')
#连接kafka的集群(可以不是集群),以及定义group id,auto_offset_reset:重置偏移量,earliest移到最早的可用消息,latest最新的消息,默认为latest
s=['test'] #指定topic,可以是多个
consumer.subscribe(s) #订阅topic
while 1:
raw_message=consumer.poll(timeout_ms=1000,max_records=10000) #设置超时时间
for topic_patition,message in raw_message.items():
for i in message:
print (i.topic,i.value) #打印出这条消息以及他的topic
生产者:
from kafka import KafkaProducer
producer=KafkaProducer(bootstrap_servers=['192.168.0.0','192.168.0.1','192.168.0.2']) #指定生产集群
producer.send('test','message')#向集群中名字为'test'的topic发送'message'
producer.flush() #冲刷缓存区,保证消息发送