如何用asyncio实现异步入队、同步处理事件？遇await报错求解

你的核心问题是每次调用event都创建新的处理任务，导致多个任务同时消费队列（并行处理），而且任务未妥善管理引发报错。正确的做法是启动单个常驻消费者任务循环处理队列，event仅负责将事件放入队列，无需每次创建任务。

完整实现代码 #

import asyncio

class EventHandler:
    def __init__(self):
        self._queue = asyncio.Queue()
        self._loop = asyncio.get_running_loop()
        # 启动唯一的常驻消费者任务，负责逐个处理事件
        self._consumer_task = self._loop.create_task(self._consume_events())

    def event(self, flag, *args):
        """同步调用接口：外部可随时调用，将事件加入队列"""
        self._queue.put_nowait((flag, *args))

    async def _consume_events(self):
        """常驻任务：循环从队列取事件，按顺序逐个处理"""
        try:
            while True:
                # 阻塞等待队列中的事件
                flag, *args = await self._queue.get()
                try:
                    # 执行具体事件处理逻辑
                    await self._process_event(flag, *args)
                finally:
                    # 标记当前队列任务完成，避免队列积压
                    self._queue.task_done()
        except asyncio.CancelledError:
            # 任务被取消时正常退出
            pass
        except Exception as e:
            # 捕获异常防止消费者任务崩溃，可根据需求添加日志或重启逻辑
            print(f"事件处理出错: {str(e)}")
            # 可选：自动重启消费者任务
            self._consumer_task = self._loop.create_task(self._consume_events())

    async def _process_event(self, flag, *args):
        """自定义事件处理逻辑，替换为你的业务代码"""
        print(f"处理事件: flag={flag}, 参数={args}")
        # 模拟处理耗时（比如IO操作）
        await asyncio.sleep(1)

    def close(self):
        """关闭EventHandler，清理消费者任务"""
        if not self._consumer_task.done():
            self._consumer_task.cancel()
            # 同步环境下等待任务结束，异步环境可直接await self._consumer_task
            self._loop.run_until_complete(self._consumer_task)

关键说明 #

1. 同步接口无报错：event是同步函数，仅调用put_nowait将事件放入队列，无需await，外部可随时调用，不会触发未await的报错。
2. 顺序处理保证：仅启动一个_consume_events常驻任务，循环从队列取事件，确保事件按加入顺序逐个处理，不会并行执行。
3. 任务稳定性：捕获异常和取消信号，避免消费者任务意外崩溃；通过task_done和join机制，可等待所有事件处理完成后再退出。

使用示例 #

async def main():
    handler = EventHandler()
    # 模拟外部多次同步调用event
    handler.event("用户登录", "user_001")
    handler.event("发送消息", "user_001", "你好！")
    handler.event("用户登出", "user_001")
    
    # 等待队列中所有事件处理完成
    await handler._queue.join()
    # 关闭清理
    handler.close()

if __name__ == "__main__":
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者jettae schroff