在使用 confluent-kafka-python 时，首次调用 poll(0) 可能返回 None，并非因无消息，而是消费者尚未完成分区分配（assignment）——此时 consumer.assignment() 为空。需主动等待分配完成，才能可靠消费后续生产的消息。
在使用 confluent-kafka-python 时，首次调用 `poll(0)` 可能返回 `none`，并非因无消息，而是消费者尚未完成分区分配（assignment）——此时 `consumer.assignment()` 为空。需主动等待分配完成，才能可靠消费后续生产的消息。
Kafka 消费者在调用 subscribe() 后，并不会立即获得分区分配；它需先加入消费者组、触发协调器的分区分配流程（可能涉及 rebalance），这一过程异步发生且需要一定时间。即使主题已存在、仅有一个分区、且消费者是组内唯一成员，poll(0) 的零超时调用仍无法保证分配已完成——因此首次（甚至前几次）poll(0) 返回 None 是正常行为，而非配置错误或数据缺失。
正确的初始化模式是：显式等待分配完成，再进入主消费循环。推荐做法如下：
from confluent_kafka import Consumer
self.consumer = Consumer({
"bootstrap.servers": self.bootstrap_servers,
"group.id": self.group_id,
"auto.offset.reset": "latest", # 或 "earliest"，但逻辑需适配
"enable.auto.commit": False, # 建议手动控制 offset 提交
})
self.consumer.subscribe(self.topic_names, on_assign=print_assignment)
# 等待分区分配完成（阻塞直到 assignment 非空）
while not self.consumer.assignment():
self.consumer.poll(timeout=0.1) # 使用小正超时避免忙等，更高效
# 此时 assignment 已就绪，可安全 poll 消息
msg = self.consumer.poll(timeout=1.0)
if msg is not None and not msg.error():
print(f"Received: {msg.value().decode()}")
⚠️ 注意事项：
- poll(0) 在高并发或低延迟场景下易导致 CPU 空转，建议改用 poll(timeout=0.05) 或更高（如 0.1 秒），平衡响应性与资源消耗；
- 若 auto.offset.reset="earliest"，分配完成后首次 poll() 将尝试拉取最早可用消息（包括已过期消息），需确认日志清理策略是否匹配预期；
- on_assign 回调仅在分配发生时触发，但不能替代对 assignment() 的轮询检查——因为回调执行时机与 poll() 调用无严格同步保证；
- 生产环境应添加超时保护，避免无限等待（例如最多重试 30 秒）：
import time
start_time = time.time()
while not self.consumer.assignment():
self.consumer.poll(timeout=0.1)
if time.time() - start_time > 30:
raise RuntimeError("Failed to get partition assignment within 30s")
总结：Kafka 消费者“注册”到主题的本质是完成组协调与分区分配，该过程不可跳过。通过主动轮询 assignment() 并配合合理超时，可确保后续 poll() 调用具备确定性行为，彻底规避“首条消息丢失”或“多次空轮询”的常见陷阱。
