kafka-python 从消费者重新启动后上次生成的消息中读取

Question

我正在使用kafkapython来消费来自kafka队列（kafka版本0.10.2.0）的消息。特别是我使用的是KafkaConsumer类型。如果消费者停止，并且在一段时间后重新启动，我希望从最新生成的消息重新启动，即删除消费者停止时生成的所有消息。我怎样才能做到这一点？

谢谢

Answer 1

在回答您的问题时：

我的理解是，当你执行消费者.poll（）时，会返回一个字典。所以，当我想轮询信息时，我用一个循环来遍历字典。

consumer = KafkaConsumer('mytopic', bootstrap_servers=[server], group_id=group_id, enable_auto_commit=True)
messages = consumer.poll()
data = []
for msg in messages:
    for value in messages[msg]:
       #Add just the values to the list
       data.append(value[6])

我相信你正在做的是得到迭代器与消费者=Kafka消费者（'mysubject'，bootstrap_servers=[服务器]，group_id=group_id，enable_auto_commit=True），然后步行迭代器与

#start iterate
for message in consumer:
    print(message)

看起来你实际上并没有得到500个投票结果。您可以通过将max_poll_records=5添加到KafkaConsumer配置中来确认这一点。然后，当您运行代码时，如果打印出5条以上的消息，您可以知道您没有使用轮询功能。

希望有所帮助！

Answer 2

您将不会< code>seekToEnd()到日志的结尾。

请记住，您首先需要订阅主题，然后才能进行搜索。此外，订阅是懒惰的。因此，在搜索之前，您还需要添加一个“虚拟民意调查”。

consumer.subscribe(...)
consumer.poll() // dummy poll
consumer.seekToEnd()

// now enter your regular poll-loop

Answer 3

谢谢，

它起作用了！

这是我的代码的简化版本：

consumer = KafkaConsumer('mytopic', bootstrap_servers=[server], group_id=group_id, enable_auto_commit=True)
#dummy poll
consumer.poll()
#go to end of the stream
consumer.seek_to_end()
#start iterate
for message in consumer:
    print(message)

consumer.close()

文档指出poll()方法与迭代器接口不兼容，我猜想迭代器接口就是我在脚本结尾的循环中使用的接口。然而，从最初的测试来看，这段代码看起来工作正常。

使用安全吗？还是我理解错了文档？

谢谢