aiokafka的aiokafkaproducer必须显式await start()和stop()，因其为协程驱动状态机，未启动发送会报runtimeerror，不手动stop则tcp连接不释放，易触发time_wait堆积或broker连接数限制。
aiokafka 能显著提升 Kafka 的 I/O 吞吐，但直接替换 confluent-kafka 或 kafka-python 会遇到连接复用、生命周期管理和异步上下文陷阱——它不是“把 sync 换成 await”就能跑通的。
为什么 aiokafka 的 AIOKafkaProducer 必须显式 start() 和 stop()
不同于同步客户端自动管理连接，aiokafka 的生产者是协程驱动的状态机。未调用 start() 就发消息会抛出 RuntimeError: Producer is not started；不 stop() 则 TCP 连接不会释放，可能堆积 TIME_WAIT 或触发 broker 端连接数限制。
-
start()是协程，必须await，通常在async with前或__aenter__中完成 -
stop()也是协程，需显式await，不能依赖 GC 回收（它不保证调用） - 若用
async with AIOKafkaProducer(...)，框架会自动处理 start/stop，但要注意：该上下文退出后 producer 不可再用
AIOKafkaConsumer 的 getmany() vs __aiter__：吞吐与控制权的取舍
高频消费场景下，__aiter__（即 async for msg in consumer:）写法简洁，但每条消息都走一次协程调度，实际吞吐常不如批量拉取。而 getmany() 允许一次拉多条，减少网络往返和事件循环开销，适合高吞吐或需要自定义批处理逻辑的场景。
-
getmany(max_records=100, timeout_ms=100)返回Dict[TopicPartition, List[ConsumerRecord]]，需手动解析分区与消息 -
__aiter__隐式调用getone()，每条消息都触发一次 broker 请求，延迟敏感时可能放大 p99 - 注意
getmany()的timeout_ms是拉取阻塞上限，不是每条消息的超时；太小会导致空返回，太大则增加端到端延迟
如何安全地在异步 Web 服务（如 FastAPI）中复用 AIOKafkaProducer
在 FastAPI 的 lifespan hook 或全局变量中初始化单例 producer 是常见做法，但必须确保：它是协程安全的、生命周期与应用一致、且不被多个请求并发误用。
- 不要在路由函数里每次新建
AIOKafkaProducer——连接建立开销大，且未 close 会泄漏资源 - 推荐用
lifespan管理：在 startup 时await producer.start()，shutdown 时await producer.stop() - 避免在中间件或依赖注入中对 producer 做
await—— 它本身不是协程，只是提供协程方法；直接调用send()即可，后续await返回的Future - 如果需跨多个 asyncio 任务并发 send，无需额外加锁：
aiokafka内部已用asyncio.Queue序列化发送请求
aiokafka 的序列化必须自己处理，没有默认 JSON 支持
aiokafka 不像 kafka-python 提供 value_serializer 参数，所有数据都以 bytes 传输。传入非 bytes 会直接报 TypeError: a bytes-like object is required。
- 发送前手动
json.dumps(...).encode()，消费后json.loads(msg.value.decode()) - 注意字符编码：默认用
utf-8，若消息含非 UTF-8 字节（如二进制 protobuf），需用msg.value原始 bytes 处理 - 别在
send()的value参数里传None——会发空 bytes，消费者收到的是b''，不是None
真正卡住人的往往不是语法，而是 aiokafka 把连接状态、批处理时机、序列化责任全摊平给你——它高效，但拒绝隐藏复杂性。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
