Python 事件驱动架构的 Python 实现

直接报 runtimeerror: no running Event loop；必须在 asyncio.run() 内或已启动循环中调用 create_task()，模块顶层或同步函数中调用均非法。

asyncio 事件循环启动前就调用 create_task() 会怎样？

直接报 RuntimeError: no running event loop —— 不是语法错，是运行时环境缺失。python 的 asyncio 不像 Node.js 那样自动隐式启动事件循环，它必须显式进入或手动启动。

• 正确做法：所有 create_task()、run_until_complete() 必须在 asyncio.run() 内部，或在已启动的事件循环上下文中执行
• 常见错误场景：模块顶层直接写 asyncio.create_task(some_coro())，或在同步函数里漏掉 await 就返回协程对象
• 调试提示：用 asyncio.get_running_loop() 检查当前是否有活跃循环，比捕获异常更早发现问题

用 asyncio.Queue 做生产者-消费者通信，为什么消息总丢？

不是队列坏了，而是没等消费者真正处理完。默认情况下，put_nowait() 只保证入队成功，get() 返回后也不代表你 await 的逻辑已结束——漏掉 await queue.join() 是最常被忽略的环节。

• queue.task_done() 必须在每个消费者完成实际工作后显式调用，否则 join() 永远阻塞
• 不要混用 put_nowait() 和 await put()：前者可能抛 asyncio.QueueFull，后者会自动等待空闲空间，更适合事件驱动下的背压控制
• 注意大小限制：asyncio.Queue(maxsize=1) 在高并发下极易成为瓶颈，不设限（默认 0）反而更符合“事件流”语义，但需配合内存监控

想用 asyncio.Event 做跨任务信号通知，但有时不生效？

核心问题在于 set() 和 wait() 的时序竞争：如果 set() 发生在 wait() 调用之前，信号就丢失了——Event 不是消息队列，它只存一个布尔状态。

• 安全模式：总是先 await event.wait()，再由另一任务 event.set()；若无法保证顺序，改用 asyncio.Condition 或带计数的 asyncio.Semaphore
• event.is_set() 返回的是瞬时快照，不能替代 wait()；轮询检查不如挂起等待，既耗 CPU 又错过时机
• 注意作用域：同一个 Event 实例必须被所有相关任务共享，别在每次循环里新建一个

第三方库（比如 aiohttp 或 aiomysql）和自定义事件逻辑怎么共存？

它们本身都基于 asyncio，但默认共享同一个事件循环——冲突点往往出在「谁 owns 这个循环」。比如你在 asyncio.run() 外又手动调用 loop.run_forever()，或者多个框架试图安装自己的策略（如 uvloop），就会触发 RuntimeError: this event loop is already running。

立即学习“Python免费学习笔记（深入）”；

• 统一入口：整个应用只用一次 asyncio.run(main())，所有异步库都在这个主协程内启动
• 避免混合策略：不要在 asyncio.run() 中又调用 uvloop.install()；如果要用 uvloop，改用 asyncio.Runner(loop_factory=uvloop.new_event_loop)（Python 3.11+）
• 警惕阻塞调用：哪怕只有一行 time.sleep(1) 或 requests.get()，都会卡住整个事件循环——必须换成 await asyncio.sleep(1) 或 aiohttp.ClientSession

事件驱动架构真正的复杂点不在语法，而在状态流转的可见性：哪个任务在等什么、谁负责清理、超时是否可取消、错误是否传播——这些没法靠装饰器或框架自动兜底，得靠设计时就明确每条消息的生命周期。