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= 200 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)