消费后如何向Kafka主题提交消息偏移量？

Question

似乎没有提交偏移量。

我正在使用 kafka Python 包。这是我的代码

from kafka import KafkaConsumer

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

msg_pack = consumer.poll(max_records=10,timeout_ms=500,update_offsets=True)
for tp,messages in msg_pack.items():
    for message in messages:
        print("%s:%d:%d: key=%s value=%s" % (tp.topic, tp.partition,
                                             message.offset, message.key,
                                             message.value))

但问题是我总是从一开始就收到所有消息。有时我什么也得不到。你能帮帮我吗？

quickstart-events:0:52: key=None value=b'1'
quickstart-events:0:53: key=None value=b'2'
quickstart-events:0:54: key=None value=b'3'
quickstart-events:0:55: key=None value=b'4'
quickstart-events:0:56: key=None value=b'5'
quickstart-events:0:57: key=None value=b'1'
quickstart-events:0:58: key=None value=b'2'
quickstart-events:0:59: key=None value=b'3'
quickstart-events:0:60: key=None value=b'4'
quickstart-events:0:61: key=None value=b'5'

Answer 1

我总是从一开始就收到所有的消息。

这应该只发生一次，基于这些

auto_offset_reset='earliest',
enable_auto_commit=True,
group_id='my-group',

有时我没有得到任何东西

那么您可能在消费者组中没有滞后（没有要消费的记录）。您需要使用 kafka-consumer-groups --describe --group my-group 来检查延迟或偏移量是否以您期望的方式提交。

如果偏移量没有按照您期望的方式提交，那么您应该禁用自动提交并手动进行