asyncio.Server 依赖 asyncio.Protocol 处理连接生命周期，start_server 的 client_connected_cb 回调使上下文分散，建议显式继承 Protocol 并在 connection_made 中管理 transport 与 reader/writer，避免阻塞操作和错误超时处理。
asyncio.Protocol 与 asyncio.Server 的底层绑定关系
直接用 asyncio.start_server 启动服务时，你实际在操作的是 asyncio.Server 实例，它内部依赖 asyncio.Protocol 处理连接生命周期。不手动继承 Protocol，就只能靠 start_server 的 client_connected_cb 回调做连接管理——这会让请求上下文（如 client_addr、stream reader/writer）散落在回调参数里，后续做连接复用或认证拦截非常别扭。
实操建议：
- 显式定义一个类继承
asyncio.Protocol，在connection_made(self, transport)中保存transport和self._reader, self._writer = asyncio.StreamReader(), asyncio.StreamWriter(...)（更推荐用asyncio.StreamReaderProtocol配合自定义StreamReader） - 避免在
data_received中直接解析完整 RPC 消息：TCP 是字节流，消息边界需自行识别（比如前 4 字节为长度头），否则会粘包 - 不要在
connection_made里 await 任何协程——该方法必须同步返回，否则连接会被立即关闭
如何安全地在 async def handler 中调用阻塞型序列化/反序列化
像 json.loads() 或 msgpack.unpackb() 这类 CPU 密集型操作，一旦在协程中直接调用，就会阻塞整个 event loop，吞掉所有并发收益。尤其当 RPC 请求体较大（>100KB）时，单次解析可能耗时几十毫秒。
实操建议：
- 用
loop.run_in_executor(None, json.loads, data_bytes)将解析丢进默认线程池，注意传入的是原始bytes，不是str - 不要用
concurrent.futures.ProcessPoolExecutor：进程间序列化开销远超收益，且 Python 的 GIL 在纯计算场景下对json这类 C 扩展影响不大 - 如果使用
orjson，它本身是异步友好的（C 实现 + 无 GIL 释放问题），可直接 await 调用，但要注意其返回的是dict而非OrderedDict，若业务依赖键序需额外处理
asyncio.wait_for 与 timeout 的真实行为差异
asyncio.wait_for(task, timeout=5) 看似简单，但它触发超时时抛出的是 asyncio.TimeoutError，而这个异常**不会自动取消被包裹的 task**——除非你显式设置 timeout 参数为 None 并自己 handle 取消逻辑，否则那个 task 仍在后台运行，可能继续写响应、占内存、甚至引发 double-write 错误。
创建 Flask + Vue.js 全栈项目，采用单端口架构、uv 依赖管理、结构化日志和生产级前端设计。当用户要求以下操作时触发：创建新的 Flask+Vue 项目、搭建 Web 应用、构建全栈 Python+Vue 应用、初始化 Flask Web 项目、设置单端口 Flask 前后端项目。触发词：新建项目、创建项目、搭建项目、Flask项目、Vue项目。
实操建议：
- 永远搭配
try/except asyncio.TimeoutError+task.cancel()使用，且要 awaittask的 cancel 完成（await asyncio.gather(task, return_exceptions=True)） - 不要用
asyncio.wait_for包裹整个 handler 协程；应只包裹真正可能卡住的 I/O 操作（如 DB 查询、HTTP 调用），并在 handler 开头设置 soft timeout 计时器 - RPC 框架层面建议统一用
asyncio.create_task(handler(...))启动，再用asyncio.wait(..., timeout=...)管理多个任务，比嵌套wait_for更可控
为什么不能直接用 asyncio.Queue 做服务端请求队列
asyncio.Queue 看起来天然适合缓冲请求，但它的 put() 和 get() 都是协程，意味着每次入队/出队都要挂起调度。当 QPS 超过 5k，光是队列本身的 await 开销就能吃掉 10%+ CPU 时间，而且它不提供优先级、死信、重试等生产必需能力。
实操建议：
- 高频场景下改用线程安全的
queue.Queue+loop.run_in_executor包装 get/put，虽然跨线程，但实测吞吐反而高 20% - 真要异步队列，选
aioredis或arq这类带持久化和重试的方案，而不是自己基于asyncio.Queue写 dispatcher - 如果只是做限流，用
asyncio.Semaphore(100)更轻量：在 handler 开头await sem.acquire()，结尾sem.release()，避免排队等待
RPC 框架最难的从来不是“怎么发请求”，而是连接生命周期管理、超时传播一致性、以及错误发生后能否干净回收资源——这些细节不压测到 10k 连接+长连接复用场景根本暴露不出来。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
