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

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

我知道这个问题类似于this one，但我想要一个具体的解决方案。

谢谢你帮我。

Answer 1

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

要让你的consumer使用Consumer Group，你需要在构造的时候设置group_id。请参阅 docs 中的 group_id 描述：

The name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If None, auto-partition assignment (via group coordinator) and offset commits are disabled. Default: 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')

Answer 2

是否可以使用来自不同服务器的消费者。我已经在下面尝试过相同的代码，它没有从 kafka 获取任何数据。

consumer = KafkaConsumer('tet', bootstrap_servers=['192.168.1.20:9092'],
                     auto_offset_reset='earliest', enable_auto_commit=True,
                     auto_commit_interval_ms=1000, group_id=None)

注：- 当我输入错误的 ip 或端口号时，它会抛出异常。

kafka-python 消费者从偏移量开始读取（自动）

kafka-python consumer start reading from offset (automatically)

python-3.x

apache-kafka

offset

kafka-consumer-api

kafka-python