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

您收到此行为是因为您的消费者未使用消费者组.对于消费者集团,消费者将定期向卡夫卡提交(保存)其职位.这样,如果它重新启动,它将从它的最后一个承诺位置开始.

要使您的消费者使用消费者组,您需要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')