基于pykafka简单实现KAFKA消费者
1.测试环境
python 3.4
zookeeper-3.4.13.tar.gz
下载地址1:
http://zookeeper.apache.org/releases.html#download
https://www.apache.org/dyn/closer.cgi/zookeeper/
https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/
下载地址2:
https://pan.baidu.com/s/1dnBgHvySE9pVRZXJVmezyQ
kafka_2.12-2.1.0.tgz
下载地址1:
http://kafka.apache.org/downloads.html
下载地址2:
https://pan.baidu.com/s/1VnHkJgy4iQ73j5rLbEL0jw
pykafka-2.8.0.tar.gz
下载地址1:
https://pypi.org/project/pykafka/
2.问题描述
使用python-kafka类库实现kafka消费者时,发现程序有时候会自动停止消费,对一些参数进行配置后无果,换成pykafka类库实现,搞定
3.代码简单实现
#-*- encoding:utf-8 -*-
__author__ = 'shouke'
from pykafka import KafkaClient
client = KafkaClient(hosts="127.0.0.1:9092")
# 获取主题
print(client.topics)
topic = client.topics['MY_TOPIC1']
# 获取消费者
consumer = topic.get_balanced_consumer('MY_GROUP1', auto_commit_enable=True, auto_commit_interval_ms=3000)
for message in consumer:
if message is not None:
print(message.offset, message.value)
参考链接: