python - kafka-python 在消费者重启后从最后产生的消息中读取

Question

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

谢谢

Answer 1

谢谢，

有用！

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

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() 方法与迭代器接口不兼容，我猜这是我在脚本末尾的循环中使用的那个。然而，从最初的测试来看，这段代码看起来可以正常工作。

使用它安全吗？还是我误解了文档？

谢谢

Answer 2

你不会到seekToEnd()日志的末尾。

请记住，您首先需要订阅一个主题，然后才能搜索。此外，订阅是懒惰的。因此，您也需要在搜索之前添加一个“虚拟投票”。

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

// now enter your regular poll-loop

Answer 3

在回答您的问题时：

据我了解，当您执行consumer.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])

我相信您正在做的是获取迭代器，consumer = KafkaConsumer('mytopic', bootstrap_servers=[server], group_id=group_id, enable_auto_commit=True)然后使用迭代器

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

看起来您实际上并不仅仅从民意调查中获得 500 个结果。max_poll_records=5您可以通过添加到 KafkaConsumer 配置来确认这一点。然后，当您运行代码时，如果打印出超过 5 条消息，您就可以知道您没有使用轮询功能。

希望有帮助！

Answer 4

这是一种将轮询返回的所有消息放在列表中的便捷方法：

while True:
  messages = [] # Store all messages
  crs = [] # Store all consumer records
  tpd = consumer.poll(timeout_ms=60000, max_records=1)
  [ crs.extend(tp) for tp in tpd.values() ] # List of cr's
  [ messages.extend([json.loads(cr.value)]) for cr in crs ]
  print messages