高吞吐 kafka 消费必须用 getmany() 批量拉取并调优 fetch 参数，配合异步处理与显式提交；producer 必须 await start()/stop() 或用 lifespan 管理；阻塞操作会导致 rebalance。
aiokafka.getmany() 必须替代 async for
async for msg in consumer: 看起来简洁，但底层每次迭代都调用 getone()，单条拉取 + 单次网络往返，吞吐直接被压到 1/5 以下。高吞吐场景下，它不是“写法问题”，而是架构级瓶颈。
正确做法是用 getmany() 批量拉取：
-
getmany(max_records=100, timeout_ms=100)——timeout_ms是拉取阻塞上限，不是单条超时；设太小（如 10ms）会导致空返回频繁，设太大（如 2000ms）会拖慢端到端延迟 - 返回值是
Dict[TopicPartition, List[ConsumerRecord]]，天然支持按分区聚合、并发处理或异步分发 - 必须在循环内显式调用
consumer.commit()或commit_async()，否则偏移量不提交，重启后重复消费
producer.start() 和 stop() 不 await 就等于没写
常见错误是把同步客户端习惯带进异步代码：在 FastAPI 路由里 new 一个 AIOKafkaProducer，直接 await producer.send(...) —— 这会立刻抛出 RuntimeError: Producer is not started。
原因很简单：start() 和 stop() 都是协程，不是普通方法：
- 漏掉
await producer.start()→ 发送失败，报错明确但容易忽略 - 漏掉
await producer.stop()→ TCP 连接不释放，高并发下快速堆积TIME_WAIT，Broker 端连接数很快打满 - 用
async with AIOKafkaProducer()可自动管理，但退出后实例不可复用，不适合长生命周期服务
推荐方式：全局单例 + FastAPI lifespan，在 startup 里 await producer.start()，shutdown 里 await producer.stop()。
fetch 参数不调优，asyncio 再快也白搭
Python 的 GIL 和 Kafka 客户端实现决定了：光靠堆 asyncio 任务数量无法线性提吞吐。真正起作用的是 fetch 层参数与业务处理节奏的匹配。
关键三项必须一起看：
-
fetch_min_bytes：默认 1 字节，意味着有消息就拉——高频小消息下网络开销爆炸。建议设为1024 * 1024（1MB），让 Broker 等够数据再响应 -
fetch_max_wait_ms：默认 500ms，和fetch_min_bytes配合使用；若 100ms 内凑不够 1MB，就直接返回当前已有的 -
max_partition_fetch_bytes：单分区单次最大拉取量，必须 ≤ Broker 的message.max.bytes，否则请求被拒
这三个值不协调，getmany() 就拉不到预期批量，后续所有异步处理都成空转。
业务逻辑阻塞 poll 循环，心跳一断就 rebalance
消费者维持组内存活靠心跳，心跳由后台线程发出，但前提是 poll 循环不能卡住。一旦你在 getmany() 拿到消息后，直接在同一线程里做耗时操作（比如调用 Playwright 同步 API、复杂 JSON 解析、数据库写入），poll 就会停摆。
结果就是：心跳超时 → Broker 认为消费者死亡 → 触发 rebalance → 分区重分配 → 消费暂停几秒甚至几十秒。
解法只有两个：
- 耗时操作必须交出控制权：用
loop.run_in_executor()托管到线程池，或改用原生异步库（如playwright.async_api） - 避免在 consumer 实例上做任何阻塞调用；所有处理应以 task 形式提交，保持 poll 循环始终可调度
最容易被忽略的是：rebalance 不是“偶尔发生”，而是一旦发生，整个消费者组都会短暂失能。它不像错误日志那样显眼，但会悄悄吃掉你的 SLA。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
