Python asyncio 超时与取消：避免后台任务越过业务生命周期

一次业务操作同时启动多个协程时，需要明确超时后谁取消子任务、谁等待资源清理、谁处理最终异常。本文所说的“任务泄漏”，指任务越过原本的业务生命周期继续运行，并不等同于所有内存泄漏问题。

例如接口请求已经超时，关联任务却仍在等待网络响应，反复重试可能累积后台工作。检查时应沿任务创建位置确认其归属和退出路径。

三种等待方式的差别

asyncio.wait(..., timeout=...) 返回已完成和未完成任务集合，超时不会自动取消后者。asyncio.wait_for() 在超时时取消目标，并等待取消过程完成，因此实际等待可能超过设定时长。

对于生命周期一致的一组子任务，Python 3.11 及以上可以用 TaskGroup 管理，在外层设置 asyncio.timeout()。任务组负责等待组内任务退出；超时上下文通过取消当前任务触发超时，并在退出时转换为 TimeoutError，所以应在上下文外捕获。

一个不访问网络的完整示例

以下程序无第三方依赖，模拟两个长期等待的任务。保存为 timeout_demo.py 后，使用 Python 3.11 及以上执行。

import asyncio


async def worker(name, cleaned):
    try:
        await asyncio.sleep(3600)
    finally:
        # 模拟需要 await 的资源清理
        await asyncio.sleep(0)
        cleaned.add(name)


async def main():
    cleaned = set()
    tasks = []

    try:
        async with asyncio.timeout(0.05):
            async with asyncio.TaskGroup() as group:
                for name in ("A", "B"):
                    task = group.create_task(worker(name, cleaned))
                    tasks.append(task)
    except TimeoutError:
        print("操作超时")

    print("已清理:", ",".join(sorted(cleaned)))
    print("子任务全部结束:", all(t.done() for t in tasks))
    print("子任务全部取消:", all(t.cancelled() for t in tasks))


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

预期输出如下，示例已在 Python 3.11.16 实际运行核对：

操作超时
已清理: A,B
子任务全部结束: True
子任务全部取消: True

清理结果排序后输出，避免依赖两个任务的退出顺序。检查发生在 main() 返回前，可以确认清理已经完成，而不是依靠 asyncio.run() 结束时才处理遗留任务。

应用到业务代码时的边界

可将清理逻辑替换为连接关闭、临时资源释放等操作。CancelledError 继承自 BaseException；显式捕获它时，通常应在清理后重新抛出，吞掉取消可能破坏超时与任务组的协作。

取消是协作式的。长时间同步阻塞会让事件循环无法及时处理取消，清理耗时也会使退出延后；它不能强制终止线程，也不保证严格的墙钟时间上限。清理若抛出新异常，应保留分析，不能全部当作普通超时忽略。

确认子任务通过 group.create_task() 创建。仅将代码放入 TaskGroup，不会自动接管其他地方调用 asyncio.create_task() 创建的任务。确需跨请求存活的后台工作，应单独定义所属组件、异常处理和关闭流程。

本地取消也不代表远端操作撤销：请求可能已被服务端接收，超时重试仍应结合幂等设计。排查时依次回答：任务由谁创建、业务结束时由谁停止、退出后由谁确认结果。

参考资料

Python 3.14 标准库官方文档：asyncio — Coroutines and Tasks 中的 Waiting Primitives、Task Groups、Timeouts，以及 asyncio — Exceptions 中的 CancelledError。

本文由 AI 辅助整理，示例经过实际运行验证，不涉及真实业务经历或生产性能结论。