Kafka Consumer：如何从 Python 中的最后一条消息开始消费答案

【问题标题】：Kafka Consumer: How to start consuming from the last message in PythonKafka Consumer：如何从 Python 中的最后一条消息开始消费
【发布时间】：2014-08-30 22:37:06
【问题描述】：

我正在使用 Kafka 0.8.1 和 Kafka python-0.9.0。在我的设置中，我设置了 2 个 kafka 代理。当我运行我的 kafka 消费者时，我可以看到它从队列中检索消息并跟踪两个代理的偏移量。一切都很好！

我的问题是，当我重新启动消费者时，它会从头开始消费消息。我所期望的是，在重新启动时，消费者会从它在死亡之前停止的地方开始消费消息。

我确实尝试跟踪 Redis 中的消息偏移量，然后在从队列中读取消息之前调用 consumer.seek 以确保我只收到以前未见过的消息。虽然这行得通，但在部署此解决方案之前，我想与大家核实一下……也许我对 Kafka 或 python-Kafka 客户端有一些误解。似乎消费者能够从中断的地方重新开始阅读是非常基本的功能。

谢谢！

【问题讨论】：

标签： python apache-kafka kafka-consumer-api kafka-python

【解决方案1】：

卡夫卡消费者is able to store offsets in Zookeeper。在 Java API 中，我们有 two options - 高级消费者，它为我们管理状态并在重启后从离开的地方开始消费，以及没有这种超能力的无状态低级消费者。

根据我对 Python 的消费者代码 (https://github.com/mumrah/kafka-python/blob/master/kafka/consumer.py) 的理解，SimpleConsumer 和 MultiProcessConsumer 都是有状态的，并且会跟踪 Zookeeper 中的当前偏移量，所以很奇怪你有这个重复消耗的问题。

确保您在重新启动时具有相同的消费者组 ID（可能是您将其设置为随机的？）并检查以下选项：

auto_commit: default True. Whether or not to auto commit the offsets
auto_commit_every_n: default 100. How many messages to consume
                     before a commit
auto_commit_every_t: default 5000. How much time (in milliseconds) to
                     wait before commit

您可能会消耗

【讨论】：

【解决方案2】：

kafka-python 使用 kafka 服务器存储偏移量，而不是在单独的 zookeeper 连接上。不幸的是，支持提交/获取偏移量的 kafka 服务器 api 直到 apache kafka 0.8.1.1 才完全起作用。如果你升级你的 kafka 服务器，你的设置应该可以工作。我还建议将 kafka-python 升级到 0.9.4。

[kafka-python 维护者]

【讨论】：

【解决方案3】：

请注意 kafka-python 库。它有一些小问题。

如果速度对您的消费者来说不是真正的问题，您可以在每条消息中设置自动提交。它应该可以工作。

SimpleConsumer 提供了一个 seek 方法 (https://github.com/mumrah/kafka-python/blob/master/kafka/consumer/simple.py#L174-L185)，允许您在任何时候开始使用消息。

最常见的调用是：

consumer.seek(0, 0) 从队列的开头开始读取。
consumer.seek(0, 1) 从当前偏移量开始读取。
consumer.seek(0, 2) 跳过所有待处理的消息并开始只阅读新消息。

第一个参数是这些位置的偏移量。这样，如果您调用consumer.seek(5, 0)，您将跳过队列中的前 5 条消息。

另外，不要忘记，偏移量是为消费者组存储的。确保您一直使用同一个。

【讨论】：

【解决方案4】：

首先，您需要设置一个 group_id，记录偏移量，以便它将继续使用来自此 group_id 的消息。

如果您已经消费了组中所有现有的消息，那么您想再次重新消费这些消息。您可以使用seek 来实现这一点。

这是一个例子：

def test_consume_from_offset(offset):
    topic = 'test'
    consumer = KafkaConsumer(bootstrap_servers=broker_list, group_id='test')
    tp = TopicPartition(topic=topic, partition=0)
    consumer.assign([tp])
    consumer.seek(tp, offset)   # you can set the offset you want to resume from.
    for msg in consumer:
        # the msg begins with the offset you set
        print(msg)

test_consume_from_offset(10)

【讨论】：

【解决方案5】：

您只需要确保您的 Kafka Consumer 从 latest 偏移量 (auto.offset.reset="latest") 开始读取。还要确保您定义了一个消费者组，以便可以提交偏移量，并且当消费者宕机时可以选择其最后提交的位置。

使用confluent-kafka-python

from confluent_kafka import Consumer


c = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'mygroup',
    'auto.offset.reset': 'latest'
})

c.subscribe(['my_topic'])

使用kafka-python

from kafka import KafkaConsumer


consumer = KafkaConsumer(
    'my_topic', 
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='latest', 
    enable_auto_commit=True,
    group_id='mygroup'
)

【讨论】：