用python编写Kafka消费者投票消息

Question

我在使用者组中轮询来自 Kafka 的消息时遇到问题。我的使用者对象分配给给定的分区

self.ps = TopicPartition(topic, partition )

之后，消费者向该分区分配：

self.consumer.assign([self.ps])

之后，我可以计算分区内的消息

self.consumer.seek_to_beginning(self.ps)
pos = self.consumer.position(self.ps)

和self.consumer.seek_to_end（self.ps） .....

在我的主题中有超过30000条消息。问题是我只收到一条消息。

具有< code > max _ poll _ records = 200 < code > AUTO _ OFFSET _ RESET 的消费者配置是最早的

这是我的函数，我正在尝试获取消息：

 def poll_messages(self):


    data = []

    messages = self.consumer.poll(timeout_ms=6000)


    for partition, msgs in six.iteritems(messages):

        for msg in msgs:

            data.append(msg)

    return data

即使我在开始轮询消息之前转到第一个可用的偏移量，我也只会收到一条消息。

self.consumer.seek(self.ps, self.get_first_offset())

我希望有人能解释我做错了什么。提前感谢。

最好的祝愿Jörn

Answer 1

我相信你误解了max_poll_records——这并不意味着你每次投票会得到200条，只是你可能得到的最多的一个限制。您需要多次调用轮询。我可以参考文档中的简单示例：http://kafka-python.readthedocs.io/en/master/usage.html

我认为更标准的实现是：

for message in self.consumer:
  # do stuff like:
  print(msg)