asyncio Future无法解析：后台协程事件发射器回调调用set_result无效

你猜的完全对——asyncio.Future和创建它的事件循环绑定死了，要是在别的事件循环或者非asyncio线程里调用set_result()，哪怕触发了这个方法，等待的await也会一直挂着。因为asyncio的事件循环根本没察觉到这个操作，没法唤醒等待的任务。

核心原则：在创建Future的事件循环里操作它 #

不管你的end_handler是在别的线程还是别的事件循环里触发的，都得把set_result()的操作“递”到创建Future的那个事件循环里执行：

1. 线程场景：用loop.call_soon_threadsafe #

如果end_handler是WebSocket库在后台线程触发的回调，直接用事件循环的线程安全方法调用set_result：

# 假设你在创建Future时拿到了对应的事件循环
loop = asyncio.get_running_loop()
# 在end_handler里：
if future and not future.done():
    loop.call_soon_threadsafe(future.set_result, collected_data)

2. 跨事件循环场景：用队列传递信号 #

如果是在另一个asyncio事件循环里处理的end信号，就用asyncio.Queue把结果传到原循环：

# 在客户端初始化时创建跨循环队列
self.result_queue = asyncio.Queue()

# 原循环里启动一个监听队列的任务
async def _queue_listener():
    while True:
        msg_id, data = await self.result_queue.get()
        future = self.pending_futures.pop(msg_id, None)
        if future and not future.done():
            future.set_result(data)

# 在另一个事件循环的end_handler里：
await self.result_queue.put((msg_id, collected_data))

3. 修正后的客户端关键代码示例 #

这里给你一个能支持多任务并发的客户端核心实现，解决Future挂起的问题：

import asyncio
import websockets
import uuid

class WSConcurrentClient:
    def __init__(self):
        self.ws = None
        self.loop = asyncio.get_running_loop()
        self.pending_tasks = {}  # 用唯一ID映射Future和数据缓存

    async def connect(self, server_url):
        self.ws = await websockets.connect(server_url)
        # 启动后台消息监听任务（同事件循环）
        asyncio.create_task(self._listen_for_responses())

    async def _listen_for_responses(self):
        async for raw_msg in self.ws:
            msg_type, msg_id, content = raw_msg.split("|", 2)
            if msg_type == "data":
                # 收集响应片段
                if msg_id in self.pending_tasks:
                    self.pending_tasks[msg_id]["data"].append(content)
            elif msg_type == "end":
                # 触发Future解析
                if msg_id in self.pending_tasks:
                    future = self.pending_tasks[msg_id]["future"]
                    collected_data = self.pending_tasks.pop(msg_id)["data"]
                    # 确保在当前循环（创建Future的循环）设置结果
                    if not future.done():
                        self.loop.call_soon_threadsafe(future.set_result, collected_data)

    async def send_request(self, request_data):
        # 生成唯一任务ID，支持多任务并发
        task_id = str(uuid.uuid4())
        # 创建绑定到当前循环的Future
        future = self.loop.create_future()
        # 缓存Future和数据列表
        self.pending_tasks[task_id] = {"future": future, "data": []}
        # 发送带任务ID的请求
        await self.ws.send(f"req|{task_id}|{request_data}")
        # 等待结果返回
        return await future

关键检查点 #

• 打印future._loop和asyncio.get_running_loop()，确认是否为同一个对象，不一样就会出问题
• 多任务场景下必须用唯一ID区分每个请求的Future，不能共用
• 后台监听任务必须是asyncio.create_task创建的asyncio任务，不能用线程池的线程

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