Kafka-Python 消费者从偏移量开始读取(自动)

Question

我正在尝试使用 kafka-python 构建一个应用程序，其中消费者从一系列主题中读取数据。极其重要的是，消费者永远不会两次阅读相同的消息，但也永远不会错过任何消息。

一切似乎都工作正常，除了当我关闭消费者(例如失败(并尝试从偏移量开始读取时。我只能读取主题中的所有消息(这会创建双重读取(或仅侦听新消息(并错过在故障期间发出的消息(。暂停消费者时我没有遇到这个问题。

我创建了一个孤立的模拟来尝试解决问题。

这里是通用生产者：

from time import sleep
from json import dumps
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
x=0 # set manually to avoid duplicates 
for e in range(1000):
if e <= x:
pass
else:
data = dumps(
{
'number' : e
}
).encode('utf-8')
producer.send('numtest', value=data)
print(e, ' send.')
sleep(5)

还有消费者。如果auto_offset_reset设置为'earliest'，将再次读取所有消息。如果auto_offset_reset设置为'latest'，则在停机期间不会读取任何消息。

from kafka import KafkaConsumer
from pymongo import MongoClient
from json import loads
## Retrieve data from kafka (WHAT ABOUT MISSED MESSAGES?)
consumer = KafkaConsumer('numtest', bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest', enable_auto_commit=True,
auto_commit_interval_ms=1000)

## Connect to database
client = MongoClient('localhost:27017')
collection = client.counttest.counttest
# Send data
for message in consumer:
message = loads(message.value.decode('utf-8'))
collection.insert_one(message)
print('{} added to {}'.format(message, collection))

我觉得自动提交无法正常工作。

我知道这个问题与这个问题相似，但我想要一个具体的解决方案。

谢谢你帮助我。

Answer 1

您出现此行为是因为您的使用者未使用使用者组。对于消费者组，消费者将定期将其位置提交(保存(给 Kafka。这样，如果重新启动，它将从上次提交的位置开始。

要使您的使用者使用使用者组，您需要在构造它时设置group_id。 请参阅文档中group_id说明：

要加入动态分区的消费组的名称 赋值(如果已启用(，并用于获取和提交 补偿。如果为 None，则自动分区分配(通过组协调器( 并且禁用偏移提交。默认值：无

例如：

consumer = KafkaConsumer('numtest', bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest', enable_auto_commit=True,
auto_commit_interval_ms=1000, group_id='my-group')