### Confluent Kafka Python 客户端配置 `poll` 参数 在使用 Confluent Kafka 的 Python 客户端时，可以通过设置消费者参数来控制两次 `poll` 调用之间的最大间隔以及每次 `poll` 获取的消息数量。 #### 控制两次 `poll` 调用之间的时间间隔 通过设置 `max.poll.interval.ms` 参数可以定义消费者在调用 `poll()` 方法处理消息期间允许的最大时间间隔。如果消费者的处理时间超过此值，则可能会被标记为失联并触发重新平衡。默认情况下，该值为 5 分钟 (300,000 毫秒)[^1]。 可以在初始化消费者对象时传递此参数： ```python from confluent_kafka import Consumer conf = { 'bootstrap.servers': 'your_broker_address', 'group.id': 'your_group_id', 'auto.offset.reset': 'earliest', # 或其他偏移量重置策略 'max.poll.interval.ms': 60000 # 设置两次 poll 间最大时间为 60 秒 } consumer = Consumer(conf) ``` #### 控制每次 `poll` 返回的消息数量 通过设置 `fetch.min.bytes` 和 `fetch.max.wait.ms` 可间接影响每次 `poll` 批次返回的消息数量。然而，更直接的方式是利用 `max.poll.records` 参数限制每次 `poll` 调用最多返回的消息条数，默认值为 500 条记录[^2]。同样，在创建消费者实例时可指定此参数： ```python conf = { 'bootstrap.servers': 'your_broker_address', 'group.id': 'your_group_id', 'auto.offset.reset': 'earliest', 'max.poll.records': 100 # 每次 poll 最大返回 100 条消息 } consumer = Consumer(conf) # 订阅主题 consumer.subscribe(['your_topic']) try: while True: msg = consumer.poll(timeout=1.0) # timeout 单位为秒 if msg is None: continue elif msg.error(): print(f"Consumer error: {msg.error()}") else: print(f"Received message: {msg.value().decode('utf-8')}") finally: consumer.close() ``` 以上代码片段展示了如何配置 `max.poll.records` 和执行基本的消费逻辑。 --- ### 注意事项 - 如果业务场景中存在较长时间的消息处理需求，建议适当增加 `max.poll.interval.ms` 值以避免因超时引发不必要的再均衡事件。 - 同样需要注意的是，增大批次大小 (`max.poll.records`) 将提升内存占用率；因此需依据实际硬件资源合理调整这些参数。 ---