应避免使用 async for，改用 getmany() 批量拉取消息以提升吞吐量；其性能更高因减少网络往返与协程调度开销，并需合理配置 fetch 参数、显式提交偏移量、防止 event loop 阻塞及内存引用问题。
直接用 async for msg in consumer: 处理高吞吐消息，吞吐量通常只有 getmany() 的 30%～50%，根本原因不是代码写得“不够异步”，而是底层每次迭代都触发一次 getone() 网络拉取+事件循环调度，开销被放大了。
别用 async for，改用 getmany() 批量拉取
async for 表面简洁，实际每条消息都走独立的 fetch 请求和协程调度，高频场景下网络往返和 asyncio 唤醒成本压倒业务逻辑。而 getmany() 一次从多个分区批量拉取，返回 Dict[TopicPartition, List[ConsumerRecord]]，天然适配批处理、并发解析或异步转发。
-
getmany(max_records=100, timeout_ms=100)中的timeout_ms是“最多等多久”，不是“每条消息超时”——若缓冲区已有数据，会立即返回，不卡住 -
max_records建议设为 50～200：太小（如 10）仍频繁调用；太大（如 1000）可能让单次处理耗时超标，触发心跳超时 - 务必在
getmany()后显式调用consumer.commit()或consumer.commit_async()，否则偏移量不会更新
fetch 参数必须协同调优，不能只改 max_poll_records
Python 的 aiokafka 不支持 max.poll.records（那是 kafka-python 的参数），它对应的是 fetch_max_wait_ms 和 fetch_min_bytes。这两个值不匹配，getmany() 就无法攒够批次。
-
fetch_min_bytes=1048576（1MB） +fetch_max_wait_ms=100是高吞吐常见组合：避免空轮询，又不让延迟飙升 -
max_partition_fetch_bytes必须 ≥ 单条消息最大体积 ×max_records，否则getmany()会截断返回，你以为拉了 100 条，其实只拿到 3 条 - 如果消费者组里有多个实例，还要确认
session.timeout.ms≥ 业务单批处理耗时 × 2，否则会频繁 rebalance
消费逻辑不能阻塞 event loop，否则心跳失效
getmany() 拿到消息后，若直接在同一个协程里做 JSON 解析、DB 写入或 HTTP 调用，会阻塞整个 event loop —— 不仅 poll 停摆，心跳包也发不出，Broker 在 session.timeout.ms 后强制踢出该消费者，触发 rebalance。
立即学习“Python免费学习笔记（深入）”；
- CPU 密集型操作（如解密、复杂计算）必须用
loop.run_in_executor()移出 event loop - IO 密集型操作（如数据库、HTTP）必须用真正异步的库（
httpx.AsyncClient、asyncpg），不能用requests或psycopg2 - 别在循环里直接
await producer.send()转发消息——这会让当前批次处理串行化；应先收集完所有待转发消息，再用asyncio.gather()并发发送
手动提交偏移量要防重复与丢失
启用 enable_auto_commit=False 后，commit() 的时机决定 exactly-once 还是 at-least-once。高吞吐下最容易踩的坑是：处理完一批消息、还没 commit 就崩溃了，重启后重放全部。
- 推荐在整批消息成功落地（比如 DB 写入完成 + 日志落盘）后再调用
consumer.commit() - 不要用
commit_async()配合无等待逻辑——它不保证提交成功，异常时静默失败 - 若业务允许少量重复，可在处理前先
consumer.commit()（at-most-once），但需确保幂等性 - 注意
getmany()返回的每条ConsumerRecord都带offset和partition，可精确提交到某一分区某一位点，不必整批推进
最易被忽略的一点：getmany() 返回的 List[ConsumerRecord] 是浅拷贝，内部 value 和 key 仍是原始字节引用。如果你在多处并发修改或缓存它们，可能引发不可预测的内存错误或数据污染——需要深拷贝或立即反序列化。
