在同步函数中调用WebSockets异步send方法遇异常的解决问询

我的应用使用websockets和asyncio管理连接，send方法为异步实现：

async def send(self, message):
        logging.debug('send {}'.format(message))
        await self.websocket.send(message)

该方法在异步线程中使用正常，但从同步方法调用时，尝试通过asyncio.run(ws.send(json.dumps(payload)))调用出现如下异常：

Task <Task pending name='Task-134' coro=<WebSocketCommonProtocol.send() running at /usr/local/lib/python3.8/dist-packages/websockets/legacy/protocol.py:631> cb=[_run_until_complete_cb() at /usr/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending> attached to a different loop

随后尝试获取当前事件循环调用：

loop = asyncio.get_event_loop()
asyncio.run(ws.send(json.dumps(payload)), loop=loop)

却报错：

There is no current event loop in thread 'Thread-37'.

核心原因 #

WebSocket对象绑定了它被创建时所在的事件循环，asyncio.run()会创建全新的事件循环，导致协程与WebSocket所属loop不匹配；另外你在非事件循环线程中调用asyncio.get_event_loop()，自然找不到当前运行的loop。

正确实现步骤 #

1. 保存WebSocket所属的事件循环
   在初始化WebSocket连接的代码中，保存它所在的事件循环对象：

   async def connect(self, url):
       self.websocket = await websockets.connect(url)
       # 保存当前运行的事件循环
       self.loop = asyncio.get_running_loop()
   

2. 在同步方法中调用异步send
   使用asyncio.run_coroutine_threadsafe()将协程提交到WebSocket所属的事件循环中执行，这个函数专门用于跨线程向指定事件循环提交协程：

   # 同步方法中的调用代码
   import json
   import asyncio
   
   payload = {"key": "example_value"}
   # 提交协程到WebSocket绑定的loop，返回线程安全的Future对象
   future = asyncio.run_coroutine_threadsafe(ws.send(json.dumps(payload)), ws.loop)
   
   # 可选：等待发送完成并处理结果/异常
   try:
       # result()会阻塞直到协程执行完成
       future.result()
   except Exception as e:
       print(f"消息发送失败: {e}")
   

其他场景处理 #

• 如果同步代码和WebSocket的事件循环在同一个线程，且当前loop处于运行状态：
  • 无需等待结果时，直接用asyncio.create_task(ws.send(json.dumps(payload)))提交任务即可。
  • 需要等待结果时，建议将同步方法改为异步，直接使用await ws.send(...)；若无法改异步，可使用ws.loop.run_until_complete(ws.send(json.dumps(payload)))，但需确保当前loop未被阻塞。

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