Python异步编程中WebSocket监听因while True冻结问题求助

你的核心问题在于：监听函数的无限循环阻塞了事件循环，即使改用asyncio.create_task仍未解决，本质是主循环的执行逻辑没有彻底和监听任务解耦，同时缺乏异常排查机制。

具体修改方案 #

1. 彻底解耦主循环与事件处理 #

修改run方法，不再等待on_event执行完成，直接将事件处理逻辑包装为独立任务，让主循环专注于接收新消息：

async def run(self):
    async with websockets.connect(WS_URL, max_size=None) as ws:
        await self.__on_open(ws)
        while True:
            event = json.loads(await ws.recv())
            # 直接创建任务，不阻塞主循环
            asyncio.create_task(on_event(event))

2. 完善事件处理函数的异常监控 #

为创建的监听任务添加异常回调，避免任务静默失败导致的隐性问题：

# 全局定义监听列表，确保装饰器可访问
MESSAGE_LISTENERS = []

async def on_event(event: dict):
    msg = getMessage(event['data'])
    for func in MESSAGE_LISTENERS:
        task = asyncio.create_task(func(msg))
        # 添加异常处理回调
        def handle_task_error(task):
            if exc := task.exception():
                print(f"监听任务执行失败: {str(exc)}")
        task.add_done_callback(handle_task_error)

3. 优化监听函数的鲁棒性 #

在监听函数中添加异常捕获，尤其是任务取消信号，避免异常扩散影响主循环：

@on_message
async def onMessage(msg: Message):
    try:
        while True:
            print(f"Message '{msg.content}' was sent by the server")
            await asyncio.sleep(3)
    except asyncio.CancelledError:
        # 任务被取消时正常退出
        pass
    except Exception as e:
        print(f"监听函数出错: {str(e)}")

4. 处理耗时同步函数（可选） #

如果getMessage是耗时的同步函数，用asyncio.to_thread包装，避免阻塞事件循环：

async def on_event(event: dict):
    # 耗时同步函数转异步执行
    msg = await asyncio.to_thread(getMessage, event['data'])
    for func in MESSAGE_LISTENERS:
        task = asyncio.create_task(func(msg))
        task.add_done_callback(lambda t: print(f"任务错误: {t.exception()}") if t.exception() else None)

关键原理说明 #

• asyncio.create_task会将异步函数加入事件循环的任务队列，后台独立执行，不会阻塞当前代码流程
• 主循环仅负责接收消息并分发任务，不再等待任务执行完成，确保新消息能被及时处理
• 异常回调和捕获机制可以快速定位任务执行中的问题，避免隐性阻塞

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