不挂起就无法获取KafkaConsumer消息答案

【问题标题】：Cannot get KafkaConsumer messages without hanging不挂起就无法获取KafkaConsumer消息
【发布时间】：2017-07-16 06:52:41
【问题描述】：

我已经使用KafkaProducer 将一些数据写入我的单分区主题，我正在尝试通过循环通过消费者或poll() 使用KafkaConsumer 查看这些数据

import time
from datetime import datetime, timedelta
from kafka import KafkaProducer, KafkaConsumer, TopicPartition

consumer = KafkaConsumer(bootstrap_servers='localhost:9092',group_id='my-group',enable_auto_commit=False)
tp = TopicPartition(topic_name, 0)
consumer.assign([tp])
consumer.seek_to_end(tp)
last_offset = consumer.position(tp)
producer = KafkaProducer(bootstrap_servers='localhost:9092')

stopWriting = datetime.now() + timedelta(seconds=10)
while datetime.now() < stopWriting:
    producer.send(topic='my-topic',value=str(datetime.now()).encode('utf-8'))
    time.sleep(1)

producer.close()
consumer.seek(tp, last_offset)

#looping through the consumer
for msg in consumer:
    print(msg)

# or looping through the polled messages 
for msg in consumer.poll():
    print(msg)

两者似乎都无法正常工作，消费者循环确实打印出消息，但总是以kafka/consumer/group.py(886)_message_generator 内的无限循环挂起。轮询循环根本不打印任何内容。在不挂起程序的情况下读出所有新制作的消息，我是否缺少一些东西？我正在使用 Python 3.6.1 和 kafka-python 版本 1.3.4

【问题讨论】：

尝试循环执行consumer.poll()。这是一个很好的参考：confluent.io/blog/…
该网站没有太大帮助，因为没有发现异常。但坚持使用poll() 是一个不错的方向。

标签： python apache-kafka kafka-consumer-api

【解决方案1】：

我找到了poll() 的方法，首先你需要超时，因为缓冲区中没有任何消息。接下来返回一个{TopicParition:[ConsumerRecord]} 的字典，因此您需要指定要从中读取消息的主题分区。

import sys
records = consumer.poll(timeout_ms=sys.maxsize)
for record in records[tp]:
    print(record)

【讨论】：