取消asyncio.future后触发回调引发异常问题

我来帮你捋捋这个问题——在asyncio里取消gather返回的GatheringFuture确实容易踩坑，尤其是涉及到多个客户端连接的批量写入场景。结合你的代码和描述，咱们一步步拆解可行的解决办法：

先搞懂异常的根源 #

你用了asyncio.gather(..., return_exceptions=True)，这个参数会让子任务的异常被收集为结果而不是直接抛出，但当你取消整个gather返回的Future时，还是会触发asyncio.CancelledError。如果直接调用future.cancel()却不捕获这个异常，它会在事件循环里冒泡成未处理异常，这就是你遇到问题的核心。

优雅取消的正确姿势 #

1. 批量取消时捕获CancelledError #

在服务器关闭的逻辑里，不要粗暴地直接取消所有Future，而是逐个取消并await，同时捕获取消异常：

async def graceful_shutdown():
    # 遍历会话集合时用copy，避免遍历中集合修改引发错误
    for session_future in connection.sessions.copy():
        try:
            # 发起取消请求
            session_future.cancel()
            # 等待取消操作完成，捕获CancelledError
            await session_future
        except asyncio.CancelledError:
            # 可以在这里记录日志，比如"客户端写入会话已优雅取消"
            pass
        except Exception as e:
            # 处理子任务可能返回的其他异常（因为你用了return_exceptions=True）
            print(f"处理会话取消时遇到异常: {str(e)}")
    # 清空活跃会话集合
    connection.sessions.clear()

2. 在Future的生命周期里做好异常处理 #

在创建和运行gather Future的地方，加上try/except/finally，确保不管是正常完成还是被取消，都能从集合里正确移除：

async def dispatch_to_connections(self, message, connections):
    writer_future = asyncio.gather(
        *[self._dispatchMessage(message, c) for c in connections],
        loop=self.loop,
        return_exceptions=True
    )
    connection.sessions.add(writer_future)
    try:
        # 等待所有子任务完成或被取消
        results = await writer_future
        # 可以在这里统一处理子任务的异常结果
        for result in results:
            if isinstance(result, Exception):
                print(f"给客户端发消息失败: {str(result)}")
    except asyncio.CancelledError:
        # 捕获取消异常，避免事件循环报错
        print("消息写入会话已被服务器关闭取消")
    finally:
        # 不管成功还是失败，都从会话集合移除
        connection.sessions.discard(writer_future)

3. 额外注意事项 #

• 确保graceful_shutdown协程是在事件循环还在运行的时候被调用的，不要等到事件循环停止后再处理取消。
• 用connection.sessions.copy()遍历集合，因为如果在遍历过程中有些Future完成并被移除，直接遍历原集合会引发RuntimeError。
• 如果你的Python版本是3.7+，可以考虑用asyncio.create_task单独管理每个_dispatchMessage任务，但对于批量等待的场景，gather还是更高效的选择。

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