本文介绍一种线程安全的混合模式：主线程可随时调用同步方法 event() 入队事件，而所有事件由独立运行的 asyncio 事件循环在后台线程中逐个、串行、阻塞式处理（即严格 FIFO + 单任务执行），避免竞态与 await 遗漏问题。
本文介绍一种线程安全的混合模式：主线程可随时调用同步方法 `event()` 入队事件，而所有事件由独立运行的 asyncio 事件循环在后台线程中**逐个、串行、阻塞式处理**（即严格 fifo + 单任务执行），避免竞态与 await 遗漏问题。
在实际系统（如 GUI 应用、信号监听器或插件式事件总线）中，常需满足两个关键约束：
- 入队必须是同步的：外部代码（可能运行在主线程、回调函数或非 asyncio 环境）应能直接调用 event(...)，不关心协程、无需 await；
- 处理必须是串行且可控的：事件需严格按到达顺序、一个接一个地执行，且每个处理逻辑可含异步 I/O（如 API 调用、数据库写入），但整体调度由统一事件循环管理。
直接在主线程混用 async def event() 会导致 RuntimeWarning: coroutine 'event' was never awaited；而若在同步 event() 中 create_task() 后不 await，则任务虽被调度，却无法保证“串行”——多个 _event() 协程会并发争抢队列，违背“一次只处理一个”的需求。
✅ 正确解法是分离执行上下文：
- 主线程保持同步，仅负责安全投递（通过 asyncio.run_coroutine_threadsafe）；
- 另起一个专用线程托管 asyncio.EventLoop，持续 await queue.get() 并顺序执行处理逻辑（天然串行）；
- 使用 asyncio.Queue 作为线程间通信桥梁——它专为跨线程协程调用设计，put()/get() 均支持 run_coroutine_threadsafe 安全调用。
以下是精简、健壮的实现：
import asyncio
from threading import Thread
class EventHandler:
def __init__(self):
self._queue = asyncio.Queue()
self._loop = asyncio.new_event_loop() # 独立事件循环
def event(self, *args, **kwargs):
"""同步接口：任意线程均可安全调用，立即返回"""
# 线程安全地向异步队列投递事件
asyncio.run_coroutine_threadsafe(
self._queue.put((args, kwargs)),
self._loop
)
async def _process_one(self, args, kwargs):
"""单个事件的处理逻辑（可含 await）"""
print(f"Processing: args={args}, kwargs={kwargs}")
# ✅ 此处可安全使用 await：如 await asyncio.sleep(1)
# await some_async_api_call(*args, **kwargs)
async def _consumer_loop(self):
"""永续消费队列，严格串行处理"""
while True:
args, kwargs = await self._queue.get()
try:
await self._process_one(args, kwargs)
finally:
self._queue.task_done() # 标记完成，支持 join()
def start(self):
"""启动后台事件循环线程"""
def run_loop():
asyncio.set_event_loop(self._loop)
self._loop.run_until_complete(self._consumer_loop())
thread = Thread(target=run_loop, daemon=True)
thread.start()
return self
# 使用示例
if __name__ == "__main__":
handler = EventHandler().start()
# 模拟多处同步调用（如不同线程、信号处理器、GUI 回调）
handler.event("click", x=100, y=200)
handler.event("save", path="/tmp/data.json")
handler.event("sync", timeout=30)
# 主线程可继续做其他事（如保持 GUI 响应）
asyncio.run(asyncio.sleep(2)) # 等待处理完成（仅示例）
⚠️ 关键注意事项：
- 不要复用主线程 loop：asyncio.get_event_loop() 在非 asyncio 线程中会报错，必须显式创建新 loop 并在专属线程中 set_event_loop()；
- 避免 put_nowait() 跨线程调用：asyncio.Queue 的同步方法（如 put_nowait）不是线程安全的，必须通过 run_coroutine_threadsafe 调用协程版 put()；
- 处理逻辑需 await：_process_one() 必须是 async def，才能在 await self._queue.get() 后自然挂起，让出控制权给其他协程（如日志、监控任务）；
- 异常隔离：建议在 try/finally 中包裹处理逻辑，确保 task_done() 总被调用，防止 queue.join() 永久阻塞；
- 优雅退出（进阶）：可通过向队列发送 sentinel 对象（如 None）或监听 asyncio.Event 实现 stop() 方法。
该方案平衡了易用性与可靠性：对外提供纯同步接口，对内依托 asyncio 的成熟调度能力实现高并发下的确定性串行处理，是构建事件驱动中间件的推荐范式。
