python - Kafka消费者使用python进行轮询消息

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 条消息。问题是我只收到一条消息。

消费者配置:max_poll_records= 200AUTO_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())

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

最美好的祝愿乔恩

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)