aiokafka生产者必须显式await start()和stop()，否则未启动发送会报RuntimeError，不stop则TCP连接不释放，易致TIME_WAIT堆积或broker连接超限。
aiokafka 生产者必须显式 await start() 和 stop()
不调用 start() 就发消息会直接抛出 RuntimeError: Producer is not started；不调用 stop() 则 TCP 连接不会关闭，容易在高并发下触发系统 TIME_WAIT 堆积或 Broker 端连接数超限。
常见错误是把同步客户端的写法照搬过来，比如在 FastAPI 路由里每次新建 AIOKafkaProducer 并直接 send() —— 这既没 await start，又没 close，资源泄漏几乎是必然的。
-
start()和stop()都是协程，必须await，不能靠 GC 回收 - 用
async with AIOKafkaProducer(...)可自动管理生命周期，但退出后该实例不可再用 - 推荐在应用 lifespan 的
startup里await producer.start()，shutdown里await producer.stop()
getmany() 比 async for 更适合高吞吐消费
高频消费时，async for msg in consumer: 看起来简洁，但底层每条消息都走一次 getone()，网络往返和事件循环调度开销大，实际吞吐常低于批量拉取。
getmany() 允许一次从多个分区拉取最多 max_records 条消息，返回的是 Dict[TopicPartition, List[ConsumerRecord]]，更适合自定义批处理、聚合或异步并发处理。
立即学习“Python免费学习笔记（深入）”；
-
getmany(max_records=100, timeout_ms=100)的timeout_ms是拉取阻塞上限，不是单条消息超时 -
timeout_ms设太小（如 10ms）会导致频繁空返回；设太大（如 2000ms）会抬高端到端延迟 - 若业务逻辑耗时长，避免在
async for循环内直接处理，否则会阻塞 poll，导致心跳超时、rebalance
在 FastAPI 中安全复用 AIOKafkaProducer 实例
全局单例 + lifespan 管理是最常用也最稳妥的方式。关键不是“能不能共享”，而是“是否协程安全”和“生命周期是否对齐”。
直接在依赖注入或中间件中对 producer 对象本身 await 是错的——它不是协程，只是提供协程方法；你应该调用 producer.send(...)，然后 await 它返回的 Future。
- 不要在路由函数里创建新
AIOKafkaProducer：连接建立开销大，且易漏掉stop() - 多个 asyncio 任务并发调用
send()不需要额外加锁：aiokafka内部已用asyncio.Queue序列化发送请求 - 若需跨进程复用（如多 worker 模式），单例失效，应改用进程间通信或降级为同步客户端
fetch 参数与并发模型直接影响吞吐上限
Python 的 GIL 和 Kafka 客户端实现方式决定了：光靠增加线程数无法线性提升吞吐，必须配合合理的 fetch 策略和异步处理模型。
比如 fetch_min_bytes=1048576（1MB）能显著减少网络请求次数，但会提高首条消息延迟；而 max_partition_fetch_bytes 若设得比 Broker 的 message.max.bytes 小，就会被截断，浪费带宽。
-
fetch_max_wait_ms建议控制在 50–500ms 区间，与fetch_min_bytes配合压测 - 纯 CPU 密集型处理（如 JSON 解析+计算）建议用
loop.run_in_executor托管到线程池，避免阻塞 event loop - IO 密集型处理（如调用 HTTP API 或 DB）优先用
asyncio.gather并发，而非多线程
auto_offset_reset='earliest' 在重启后可能重复拉取大量历史数据；而每次消息都做 json.loads() 且未缓存解码器，会在高频下成为性能瓶颈。这些点不体现在代码行数里，却决定着系统能否稳定扛住每秒万级消息。 
