如何使用KafkaConsumer的poll()方法实现指定条数拉取与超时控制

• 核心问题是poll()方法的返回结果被完全丢弃，后续直接迭代KafkaConsumer实例本身。KafkaConsumer的迭代逻辑是持续拉取所有可用消息的无限循环，完全不受前一次poll()调用的参数约束，因此会把队列中积压的500条消息全部拉取。
• 其次要注意kafka-python的poll()方法的max_records参数仅在1.3.0及以上版本支持，同时如果初始化消费者时没有显式配置max_poll_records，默认值为500，也会导致单次拉取最多返回500条。

方案1：正确使用poll()方法获取指定数量消息 #

直接接收poll()返回的消息集合，再遍历处理即可，poll()会严格按照传入的超时时间和最大消息数返回结果：

import json
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers=kafka_server,
    group_id=consumergroup,
    client_id=consumerid,
    enable_auto_commit=False,
    auto_offset_reset='latest',
    value_deserializer=lambda m: json.loads(m.decode('ascii')),
    max_poll_records=100 # 额外配置兜底，避免版本兼容问题
)
consumer.subscribe("mytopic")

# poll返回结构为 Dict[TopicPartition, List[ConsumerRecord]]
msg_dict = consumer.poll(timeout_ms=8000, max_records=100, update_offsets=True)
all_records = []
for records in msg_dict.values():
    all_records.extend(records)

# 处理拉取到的最多100条消息
for message in all_records:
    print("offset", message.offset)

# 处理完后手动提交offset（因为关闭了自动提交）
consumer.commit_sync()

方案2：迭代消费者时手动控制拉取数量 #

如果习惯用迭代的方式消费，可以初始化时设置max_poll_records=100，迭代时取满100条就终止循环：

import json
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers=kafka_server,
    group_id=consumergroup,
    client_id=consumerid,
    enable_auto_commit=False,
    auto_offset_reset='latest',
    value_deserializer=lambda m: json.loads(m.decode('ascii')),
    max_poll_records=100, # 单次拉取最大条数
    consumer_timeout_ms=8000 # 无消息时最多等待8秒就抛出超时异常
)
consumer.subscribe("mytopic")

max_count = 100
count = 0
try:
    for message in consumer:
        print("offset", message.offset)
        count +=1
        if count >= max_count:
            break
except StopIteration:
    # 8秒无消息触发超时，结束消费
    pass

consumer.commit_sync()

内容的提问来源于stack exchange，提问作者santanu de