安装
pip3 install setuptools
pip3 install pykafka
pip3 install kafka-python
单台kafka
import sys from kafka import KafkaConsumer from kafka.structs import TopicPartition consumer = KafkaConsumer( enable_auto_commit=False, bootstrap_servers=["xxx-xx-x-sh-yyy-Eyy.yd.com:1234"]) consumer.subscribe(topics=(‘indata_str_abcde_click_new‘)) if sys.getdefaultencoding() != ‘utf-8‘: reload(sys) sys.setdefaultencoding(‘utf-8‘) for message in consumer: print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value.decode()))
kafka集群
bootstrap_servers 参数使用‘‘,‘‘,‘‘隔开
import sys from kafka import KafkaConsumer from kafka.structs import TopicPartition consumer = KafkaConsumer( enable_auto_commit=False, bootstrap_servers=["xxx-xx-x-sh-yyy-Eyy.yd.com:1234"，"xxx-xx-x-sh-yyy-Eyy.yd.com:1231"，"xxx-xx-x-sh-yyy-Eyy.yd.com:1232"]) consumer.subscribe(topics=(‘indata_str_abcde_click_new‘)) if sys.getdefaultencoding() != ‘utf-8‘: reload(sys) sys.setdefaultencoding(‘utf-8‘) for message in consumer: print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value.decode()))
可成功拿到消息
原文：https://www.cnblogs.com/jxba/p/13189886.html