本文介绍一种安全、可靠的模式：主线程可随时调用同步方法 event() 入队事件，而事件由独立的 asyncio 事件循环在后台线程中逐个、串行、异步地处理，避免竞态、阻塞和 RuntimeWarning: coroutine was never awaited 等常见错误。
本文介绍一种安全、可靠的模式：主线程可随时调用同步方法 event() 入队事件，而事件由独立的 asyncio 事件循环在后台线程中逐个、串行、异步地处理，避免竞态、阻塞和 runtimewarning: coroutine was never awaited 等常见错误。
在构建事件驱动系统（如传感器回调、GUI事件、消息总线监听器）时，常面临一个核心矛盾：事件触发是同步且不可控的（如来自主线程的频繁调用），但处理逻辑可能需要异步能力（如 await HTTP 请求、数据库操作），同时又必须严格串行执行以保证状态一致性。直接将 event() 设为 async 会导致调用方必须 await，违背“任意时刻调用”的需求；而若在同步函数中 create_task 后不妥善调度，又会触发 coroutine was never awaited 或任务丢失。
✅ 正确解法是职责分离 + 线程隔离：
- 主线程：只负责安全地将事件推入跨线程队列（asyncio.Queue），使用 asyncio.run_coroutine_threadsafe 调度；
- 专用 asyncio 线程：运行独立事件循环，持续 await queue.get() 并串行执行每个事件的处理协程（无并发风险）。
以下是完整、生产就绪的实现：
import asyncio
from threading import Thread
from typing import Any
class EventHandler:
def __init__(self):
# 创建线程安全的 asyncio.Queue（用于跨线程通信）
self._queue = asyncio.Queue()
# 创建独立事件循环（不在主线程运行）
self._loop = asyncio.new_event_loop()
def event(self, *args, **kwargs) -> None:
"""主线程安全调用：任意时刻均可触发，立即返回"""
# 使用 run_coroutine_threadsafe 安全地向另一线程的 loop 提交入队操作
asyncio.run_coroutine_threadsafe(
self._queue.put((args, kwargs)),
self._loop
)
async def _process_one_event(self, args: tuple, kwargs: dict) -> None:
"""定义具体的事件处理逻辑（可 await 任意异步操作）"""
print(f"Processing event with args={args}, kwargs={kwargs}")
# ✅ 示例：模拟异步 I/O 操作
await asyncio.sleep(0.1) # 非阻塞等待
# ... 实际业务逻辑：调用 API、写数据库、发通知等 ...
async def _event_consumer(self) -> None:
"""永续消费者：从队列取事件并串行处理"""
while True:
try:
args, kwargs = await self._queue.get()
await self._process_one_event(args, kwargs)
self._queue.task_done() # 标记完成，支持 join()
except Exception as e:
# ⚠️ 关键：捕获异常防止消费者崩溃
print(f"Error processing event: {e}")
def start(self) -> None:
"""启动后台事件循环线程"""
def run_loop():
asyncio.set_event_loop(self._loop)
# 启动单个消费者协程（确保串行）
self._loop.create_task(self._event_consumer())
self._loop.run_forever()
thread = Thread(target=run_loop, daemon=True, name="EventHandler-Loop")
thread.start()
def stop(self) -> None:
"""优雅关闭（可选）"""
self._loop.call_soon_threadsafe(self._loop.stop)
# --- 使用示例 ---
if __name__ == "__main__":
handler = EventHandler()
handler.start() # 启动后台处理线程
# 主线程中随时调用（完全同步、无 await、无阻塞）
handler.event("click", button="submit")
handler.event("timeout", duration=5000)
handler.event("data_received", payload=b'\x00\x01')
# 可选：等待所有已入队事件处理完毕
# asyncio.run_coroutine_threadsafe(handler._queue.join(), handler._loop).result()
# 程序退出前可调用 handler.stop()
? 关键设计要点说明：
- asyncio.Queue 是线程安全的：但其方法（如 put_nowait）不是线程安全的——必须通过 run_coroutine_threadsafe 在目标 loop 中调用；
- 单消费者保障串行：_event_consumer 是唯一从队列取数据的协程，天然避免并发冲突；
- 异常防护：try/except 包裹每个事件处理，防止单个失败中断整个流水线；
- Daemon 线程：daemon=True 确保主线程退出时后台线程自动终止（适合脚本/服务）；
- 可扩展性：如需多级处理或优先级队列，只需替换 _queue 为 asyncio.PriorityQueue 或添加中间协程。
此模式广泛应用于 FastAPI 后台任务、PyQt 异步桥接、IoT 设备事件总线等场景，兼顾响应性、安全性和可维护性。
