Python asyncio 超时后任务还在跑排查：从 wait_for 到取消清理

在 Python 异步项目里，超时控制看起来很简单：给远程调用包一层 asyncio.wait_for，超过时间就返回失败。但实际排查时经常会遇到一种更麻烦的现象：接口已经返回超时，后台日志却还在继续打印，连接也迟迟没有释放。

这篇文章用一个小例子把问题走一遍。我们先复现“调用方超时了，任务还在跑”的场景，再判断它和 wait_for 默认行为有什么差别，最后写出一段带取消、等待清理和结果复查的模板。

问题现场：接口已经超时，后台任务还在跑

先看一个很像线上问题的描述：接口要求 1 秒内返回，内部要调用一个慢接口。请求端看到超时提示后，服务日志里又过了几秒才打印“远程调用完成”。如果这个慢调用还占着连接池、锁或者临时状态，就可能把后续请求一起拖慢。

我们希望确认三件事：

• 超时发生后，内部任务是否真的停了。
• 如果任务还在跑，是谁阻止了取消。
• 修复后，资源清理是否在返回前完成。

初步判断：超时只是让调用方先返回了吗

很多人会把 wait_for 理解成“等一会儿，等不到就不管了”。这个理解只对一部分写法成立。要注意：直接等待一个任务时，wait_for 超时后会尝试取消它；但如果你用了 asyncio.shield，或者协程内部吞掉取消信号，任务就可能继续运行。

这也是排查时最容易漏掉的点：看到外层抛出 TimeoutError，并不等于内部任务已经结束。真正可靠的判断方式是观察任务本身的状态，并在取消后等待它走完清理逻辑。

动手验证：用日志看任务生命周期

下面这段代码故意使用 asyncio.shield，模拟“外层超时先返回，但内部任务继续跑”的场景。它常见于一些共享任务、缓存预热、批量调用封装里。

import asyncio


async def fetch_detail(order_id: int) -> str:
    print("start", order_id)
    await asyncio.sleep(5)
    print("finish", order_id)
    return f"order:{order_id}"


async def main():
    task = asyncio.create_task(fetch_detail(1001))
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout=1)
    except asyncio.TimeoutError:
        print("timeout returned to caller")

    print("task done after timeout:", task.done())
    await asyncio.sleep(6)
    print("task done later:", task.done())


asyncio.run(main())

你会看到类似输出：

start 1001
timeout returned to caller
task done after timeout: False
finish 1001
task done later: True

这说明外层超时已经返回，但任务没有被停掉。它不是 asyncio 无缘无故失控，而是 shield 明确保护了内部任务。这个结论很关键，因为修复方向不是随便加更短的超时时间，而是要决定“超时后这个任务到底应不应该继续”。

定位原因：取消信号没有真正收口

排查到这里，我们可以把常见原因分成三类：

还有一种写法也要小心：在协程里捕获 asyncio.CancelledError 后继续做耗时动作。取消信号本来是让任务尽快收尾，如果捕获后没有重新抛出，也没有明确的补偿逻辑，外层就很难知道任务到底什么时候结束。

async def risky_worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("cancel received, cleaning")
        await asyncio.sleep(1)
        raise

这段写法的重点是最后的 raise。清理可以做，但做完后要把取消状态继续交回给外层，这样上层才能正确判断任务已经被取消。

修复方案：取消任务并等待清理完成

如果业务规则是“调用方超时后，内部任务也必须停掉”，可以把创建任务、等待、取消和清理等待封装成一个小函数。注意这里不仅调用 cancel，还要再次 await 这个任务，让它有机会进入 finally 并释放资源。

import asyncio
from collections.abc import Awaitable
from typing import TypeVar

T = TypeVar("T")


async def run_with_timeout(coro: Awaitable[T], timeout: float) -> T:
    task = asyncio.create_task(coro)
    try:
        return await asyncio.wait_for(task, timeout=timeout)
    except asyncio.TimeoutError:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        raise

业务协程也要配合，把资源释放放在 finally 里：

class DemoConn:
    async def request(self, order_id: int) -> str:
        await asyncio.sleep(5)
        return f"order:{order_id}"

    async def close(self) -> None:
        print("connection closed")


async def open_conn() -> DemoConn:
    return DemoConn()


async def query_remote(order_id: int) -> str:
    conn = await open_conn()
    try:
        return await conn.request(order_id)
    finally:
        await conn.close()


async def main():
    try:
        await run_with_timeout(query_remote(1001), timeout=1)
    except asyncio.TimeoutError:
        print("timeout and cleanup done")


asyncio.run(main())

这段代码的预期输出应该先看到连接被关闭，再看到外层超时处理结束。这样调用方返回时，后台资源已经进入可控状态。

验证结果：超时、成功和外部取消三组复查

改完以后不要只跑一个超时用例，建议至少复查三组：

如果项目使用 Python 3.11 及以上，也可以在更复杂的父子任务场景里考虑 asyncio.TaskGroup，让一组任务拥有更清晰的生命周期。简单场景下，保存任务引用、超时后取消、取消后等待清理，已经能解决大部分残留任务问题。

总结清单：写 asyncio 超时控制时检查什么

• 确认超时后任务是应该停止，还是允许继续跑。
• 直接使用 wait_for 时，理解它会尝试取消被等待任务。
• 使用 shield 时，要额外处理超时后的任务去向。
• 协程里捕获 CancelledError 后，清理完成要继续抛出。
• 资源释放放进 finally，并允许外层等待清理完成。
• 由 create_task 创建的子任务要保存引用，退出前统一收口。
• 上线前同时验证正常完成、内部超时和外部取消三种路径。

这类问题的核心不是“把超时时间调短一点”，而是把任务生命周期讲清楚：谁创建任务，谁负责取消，谁等待清理完成。只要这条链路闭合，asyncio 超时后的行为就会稳定很多。