第 5 章：异步任务的超时、异常与取消

我们继续做“学生信息查询助手”。

前面已经做到两件事：

1. 可以同时查询多个学生。
2. 可以用 Semaphore 限制同一时间最多运行多少个查询。

但是项目里还有一个更常见的问题：

并发数量已经限制住了，但某个查询一直不返回，程序还是会被拖住。

比如我们模拟下面这个场景：

• 大部分学生查询 0.2 秒就能返回。
• 学生 ID 为 3 的查询特别慢，要等 10 秒。
• 有些学生 ID 在数据里不存在。
• 某些查询可能直接抛异常。

如果程序一直等慢查询，用户体验会很差。真实项目里，这种慢查询可能来自外部接口、数据库、Redis，也可能只是网络抖动。

本章要解决 4 个问题：

1. 如何给异步查询设置最大等待时间？
2. 查询超时后，如何让程序继续往下走？
3. 查询抛异常时，如何返回统一的失败结果？
4. 任务被取消时，为什么不能随便吞掉 CancelledError？

本章只讲项目够用的超时、异常和取消基础规则。不讲真实 HTTP 请求，不讲 FastAPI，不讲数据库和 Redis，也不讲复杂取消机制。

本章学习目标

学完本章，你需要能做到：

1. 解释为什么限制并发后仍然需要设置超时。
2. 使用 asyncio.timeout() 限制一段异步代码的最长等待时间。
3. 使用 try ... except TimeoutError 处理查询超时。
4. 使用 except Exception 处理普通业务异常。
5. 把成功、超时、查询不到、普通异常整理成统一结果格式。
6. 说清楚 CancelledError 的最小规则：如果捕获它，清理后要继续 raise。
7. 在已有 Semaphore 批量查询代码上增加超时和异常处理。

批量查询的慢任务场景

当前学生信息查询助手已经通过 Semaphore 限制了并发数量，能避免瞬间把数据库或外部服务冲垮。

但只限制并发数量还远远不够——如果某个查询发生网络抖动或下游服务卡死，耗时从 0.2 秒飙升到 10 秒，整个批量查询就会被这个慢任务死死拖住，前面完成的正常任务也无法提前返回。

Semaphore 只能限制同时运行的数量，不能保证每个任务都及时返回。所以本章要在查询内部再加一层超时熔断保护，并将超时、查无此人与异常统一收敛为结构化结果字典。

这一章的目标不是写复杂稳定性框架，而是让日常项目里的异步查询不再“傻等”。

查询稳定性处理

下面每一步的完整代码都直接显示在页面里。将代码按标注文件名保存到本地 _code_async/c05_timeout_and_cancellation_basics/，然后在 _code_async 目录执行命令。

1. 单次查询的超时控制

慢查询

慢查询就是某个异步操作很久没有返回。

它可能是：

1. 外部接口迟迟不响应。
2. 数据库查询太慢。
3. Redis 网络异常。
4. 第三方服务卡住。
5. 本地模拟代码里 asyncio.sleep() 时间太长。

本章用 student_id == 3 模拟慢查询。

if student_id == 3:
    await asyncio.sleep(10)

真实项目里，我们不会故意写这么慢。这里是为了让问题明显。

超时期限与作用范围

超时就是给一段异步代码设置最长等待时间。

比如：

async with asyncio.timeout(1):
    student = await query_student_async(3)

这段代码的意思是：

里面的异步操作最多等 1 秒。超过 1 秒还没完成，就进入超时处理。

日常项目里，请求外部资源时都应该考虑超时。

注意：超时不是解决所有问题。它只解决“不要一直等”。失败后要怎么提示、是否重试、是否降级，是后续更复杂的问题。本章不展开。

TimeoutError 的转换与捕获

asyncio.timeout() 超时后，外层可以捕获 TimeoutError。

达到超时期限时，超时管理器会请求取消当前任务，在可响应取消的异步等待处打断执行。退出超时块时，它把自己引发的取消转换为 TimeoutError，所以要在 async with asyncio.timeout(...) 外捕获这个错误。由外部发起的取消则应继续传播，不当作查询超时处理。

这种超时需要事件循环有机会调度任务，不能强行打断 time.sleep() 或长时间占用 CPU 的同步代码。本章查询使用 await asyncio.sleep()，能够响应取消。

超时从进入 async with asyncio.timeout(...) 时开始计时。本章批量案例先等待 Semaphore，拿到名额后才进入超时块，因此等待名额的时间不计入单次查询超时。若要限制“排队加查询”的总等待时间，应把超时块放在 Semaphore 外面。

try:
    async with asyncio.timeout(1):
        student = await query_student_async(3)
except TimeoutError:
    print("查询超时")

本章只需要记住：

1. 超时是一类失败。
2. 超时后程序不应该直接崩掉。
3. 批量任务里，某个学生超时，不应该拖住所有学生。

需求：

我们先模拟一个慢查询。学生 ID 为 3 时，查询要等待 10 秒。然后给它加 1 秒超时，让程序不要一直等。

功能分析：

1. 用 student_data 保存学生数据。
2. 用 query_student_async() 模拟异步查询。
3. 用 asyncio.timeout(1) 限制最长等待 1 秒。
4. 用 TimeoutError 捕获超时。
5. 用 time.perf_counter() 观察程序耗时。

文件名：01_timeout.py

01_timeout.py

import asyncio
import time

student_data = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}


async def query_student_async(student_id: int):
    if student_id == 3:
        await asyncio.sleep(10)
    else:
        await asyncio.sleep(0.2)

    return student_data.get(student_id)


async def main():
    start_time = time.perf_counter()

    try:
        async with asyncio.timeout(1):
            student = await query_student_async(3)
            print("查询结果：", student)
    except TimeoutError:
        print("查询超时：学生 ID 为 3 的查询超过 1 秒")

    end_time = time.perf_counter()
    print(f"总耗时：{end_time - start_time:.2f} 秒")


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

运行命令：

uv run python c05_timeout_and_cancellation_basics/01_timeout.py

可能看到的运行结果：

查询超时：学生 ID 为 3 的查询超过 1 秒
总耗时：1.00 秒

运行逻辑：

1. query_student_async(3) 会进入慢查询分支。
2. 它原本要等待 10 秒。
3. async with asyncio.timeout(1): 把这段等待限制在 1 秒内。
4. 超过 1 秒后，代码进入 except TimeoutError。
5. 程序不会等满 10 秒。

注意：

asyncio.timeout() 要包住真正的异步等待代码。这里真正要限制的是：

student = await query_student_async(3)

不要只在外面写一个空的 timeout，却没有把慢操作放进去。

2. 成功、超时与查询不到的统一结果

本阶段先把单次查询的成功、超时与查询不到整理成统一结果。普通异常会在最终批量案例中加入；当前的 02_safe_query.py 还没有处理该分支。

需求：

如果每次查询都手写 try ... except TimeoutError，代码会很散。接下来我们把查询包装成一个安全函数。

这个函数返回统一结构。

成功时：

{
    "student_id": student_id,
    "ok": True,
    "reason": "success",
    "data": student,
}

超时时：

{
    "student_id": student_id,
    "ok": False,
    "reason": "timeout",
    "data": None,
}

查询不到时：

{
    "student_id": student_id,
    "ok": False,
    "reason": "not_found",
    "data": None,
}

功能分析：

1. query_student_async() 只负责查询。
2. safe_query_student() 负责超时和结果整理。
3. 调用处只需要看统一结果，不用到处写异常处理。

文件名：02_safe_query.py

02_safe_query.py

import asyncio

student_data = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}


async def query_student_async(student_id: int):
    if student_id == 3:
        await asyncio.sleep(10)
    else:
        await asyncio.sleep(0.2)

    return student_data.get(student_id)


async def safe_query_student(student_id: int, timeout: float = 1):
    try:
        async with asyncio.timeout(timeout):
            student = await query_student_async(student_id)

        if student is None:
            return {
                "student_id": student_id,
                "ok": False,
                "reason": "not_found",
                "data": None,
            }

        return {
            "student_id": student_id,
            "ok": True,
            "reason": "success",
            "data": student,
        }

    except TimeoutError:
        return {
            "student_id": student_id,
            "ok": False,
            "reason": "timeout",
            "data": None,
        }


async def main():
    result_1 = await safe_query_student(1)
    result_3 = await safe_query_student(3)
    result_99 = await safe_query_student(99)

    print(result_1)
    print(result_3)
    print(result_99)


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

运行命令：

uv run python c05_timeout_and_cancellation_basics/02_safe_query.py

可能看到的运行结果：

{'student_id': 1, 'ok': True, 'reason': 'success', 'data': {'id': 1, 'name': '张三', 'age': 18}}
{'student_id': 3, 'ok': False, 'reason': 'timeout', 'data': None}
{'student_id': 99, 'ok': False, 'reason': 'not_found', 'data': None}

运行逻辑：

1. 查询学生 1，0.2 秒后成功返回。
2. 查询学生 3，超过 1 秒后进入超时处理。
3. 查询学生 99，查询完成，但数据不存在，所以返回 not_found。

注意：

超时和查询不到不是同一种问题。

• timeout：任务没有在规定时间内完成。
• not_found：任务正常完成了，只是没有这条数据。

批量任务里，把失败原因分清楚，后面排查会轻松很多。

3. CancelledError 的传播规则

CancelledError 表示任务被取消。

它不是普通业务失败。

asyncio.CancelledError 继承自 BaseException，不属于 Exception，因此 except Exception 不会捕获它。最终案例单独捕获取消信号并继续 raise，只把普通异常转换成失败结果。

项目够用规则很简单：

如果捕获了 asyncio.CancelledError，可以先做清理，然后继续 raise。

推荐写法：

try:
    await asyncio.sleep(5)
except asyncio.CancelledError:
    print("任务被取消，准备清理资源")
    raise

不要写成：

except asyncio.CancelledError:
    pass

这样会把取消信号吞掉，让外层以为任务正常结束。日常项目里，这很容易埋坑。

需求：

有些异步任务可能被外部取消。比如程序准备关闭、用户取消操作、上层任务不再需要结果。

这时会出现 asyncio.CancelledError。

本节只讲一个够用规则：

如果捕获 CancelledError，清理后要继续 raise。

功能分析：

1. 写一个长时间运行的查询函数。
2. 在里面捕获 CancelledError。
3. 模拟外部取消任务。
4. 观察取消结果。

文件名：03_cancel_rule.py

03_cancel_rule.py

import asyncio


async def long_running_query():
    try:
        print("开始执行长查询")
        await asyncio.sleep(5)
        print("长查询完成")
        return "查询结果"
    except asyncio.CancelledError:
        print("任务被取消，准备清理资源")
        raise


async def main():
    task = asyncio.create_task(long_running_query())

    await asyncio.sleep(0.5)
    print("外部决定取消任务")
    task.cancel()

    try:
        await task
    except asyncio.CancelledError:
        print("外层收到任务取消信号")


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

运行命令：

uv run python c05_timeout_and_cancellation_basics/03_cancel_rule.py

可能看到的运行结果：

开始执行长查询
外部决定取消任务
任务被取消，准备清理资源
外层收到任务取消信号

运行逻辑：

1. long_running_query() 开始执行，并准备等待 5 秒。
2. main() 等 0.5 秒后调用 task.cancel()。
3. 长查询内部收到取消信号，进入 except asyncio.CancelledError。
4. 函数打印清理提示，然后继续 raise。
5. 外层 await task 也能知道任务被取消了。

task.cancel() 发出取消请求，并不表示任务已经结束；外层仍需 await task，等待任务响应取消并完成退出。本例中的“准备清理资源”只是提示，代码没有实际持有或释放文件、连接等资源。

注意：

不要这样写：

except asyncio.CancelledError:
    pass

这会吞掉取消信号。外层可能以为任务正常结束，后续逻辑就容易判断错误。

本章只讲这个最小规则，不展开复杂取消传播。

4. 带超时与异常处理的批量查询

现在把前面的内容组合起来：

1. 保留并发限制。
2. 给单个查询加超时。
3. 区分成功、超时、查询不到和普通异常。
4. 对 CancelledError 做最小安全处理。
5. 批量查询后统一打印结果。

最终案例新增一个服务异常：学生 ID 为 4 时，查询函数抛出 RuntimeError("学生服务异常")。这属于普通异常，由 except Exception 转成 reason="error"；它与超时、查询不到是不同的结果。CancelledError 则继续向外传播，不转成普通失败。

批量函数保留上一章的并发值校验：limit 必须大于等于 1。若允许 limit=0，所有查询都会停在等待名额的阶段，无法进入内部的超时块；单次查询超时不能替代这个参数校验。

项目结构：

_code_async/c05_timeout_and_cancellation_basics/
├── 01_timeout.py
├── 02_safe_query.py
├── 03_cancel_rule.py
└── 04_stable_batch_query.py

安装依赖：

本章只使用 Python 标准库，不需要安装第三方依赖。

文件名：04_stable_batch_query.py

04_stable_batch_query.py

import asyncio
import time

student_data = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
    5: {"id": 5, "name": "赵六", "age": 18},
    6: {"id": 6, "name": "钱七", "age": 19},
}


async def query_student_async(student_id: int):
    if student_id == 3:
        await asyncio.sleep(10)
    elif student_id == 4:
        await asyncio.sleep(0.2)
        raise RuntimeError("学生服务异常")
    else:
        await asyncio.sleep(0.2)

    return student_data.get(student_id)


async def safe_query_student(
    student_id: int,
    semaphore: asyncio.Semaphore,
    timeout: float = 1,
):
    try:
        async with semaphore:
            async with asyncio.timeout(timeout):
                student = await query_student_async(student_id)

        if student is None:
            return {
                "student_id": student_id,
                "ok": False,
                "reason": "not_found",
                "data": None,
            }

        return {
            "student_id": student_id,
            "ok": True,
            "reason": "success",
            "data": student,
        }

    except TimeoutError:
        return {
            "student_id": student_id,
            "ok": False,
            "reason": "timeout",
            "data": None,
        }
    except asyncio.CancelledError:
        print(f"学生 {student_id} 的查询任务被取消，准备退出")
        raise
    except Exception as exc:
        return {
            "student_id": student_id,
            "ok": False,
            "reason": "error",
            "message": str(exc),
            "data": None,
        }


async def query_students_safely(student_ids, limit: int = 3, timeout: float = 1):
    if limit < 1:
        raise ValueError("limit 必须大于等于 1")

    semaphore = asyncio.Semaphore(limit)

    tasks = [
        safe_query_student(student_id, semaphore, timeout)
        for student_id in student_ids
    ]

    return await asyncio.gather(*tasks)


def format_result(result: dict) -> str:
    student_id = result["student_id"]
    reason = result["reason"]

    if result["ok"]:
        student = result["data"]
        return f"学生 {student_id} 查询成功：{student['name']}，{student['age']} 岁"

    if reason == "timeout":
        return f"学生 {student_id} 查询超时"

    if reason == "not_found":
        return f"学生 {student_id} 不存在"

    if reason == "error":
        return f"学生 {student_id} 查询失败：{result['message']}"

    return f"学生 {student_id} 查询失败：未知原因"


async def main():
    student_ids = [1, 2, 3, 4, 5, 6, 99]

    start_time = time.perf_counter()
    results = await query_students_safely(student_ids, limit=3, timeout=1)
    end_time = time.perf_counter()

    for result in results:
        print(format_result(result))

    print(f"总耗时：{end_time - start_time:.2f} 秒")


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

启动命令：

uv run python c05_timeout_and_cancellation_basics/04_stable_batch_query.py

可能看到的运行结果：

学生 1 查询成功：张三，18 岁
学生 2 查询成功：李四，19 岁
学生 3 查询超时
学生 4 查询失败：学生服务异常
学生 5 查询成功：赵六，18 岁
学生 6 查询成功：钱七，19 岁
学生 99 不存在
总耗时：1.00 秒

运行结果说明：

1. 学生 1、2、5、6 能正常查询成功。
2. 学生 3 模拟慢查询，会超时。
3. 学生 4 模拟普通异常，会返回 error。
4. 学生 99 查询完成，但数据不存在，所以返回 not_found。
5. 程序不会等待完整 10 秒。
6. limit=3 仍然控制同一时间最多 3 个查询进入受限区域。
7. 学生 4 的普通异常被整理为 ok=False, reason="error"，message 保留异常信息。
8. query_students_safely() 在创建限制器前拒绝小于 1 的并发值。

总耗时通常接近 1 秒，具体数值会随运行环境略有浮动。

关键代码说明：

async with semaphore:
    async with asyncio.timeout(timeout):
        student = await query_student_async(student_id)

这里有两层保护：

1. semaphore 控制同一时间有多少个查询在运行。
2. timeout 控制每个查询最多等多久。

注意：

Semaphore 和 timeout 解决的是两个不同问题。

• Semaphore：不要太多任务同时运行。
• timeout：单个任务不要一直等待。

不要把它们混为一谈。

本章小结

本章解决的是异步任务的基础稳定性问题。

你需要记住 4 件事：

1. Semaphore 限制并发数量，但不解决慢任务一直等待的问题。
2. asyncio.timeout() 可以限制一段异步代码的最长等待时间。
3. 批量查询时，成功、超时、查询不到、普通异常最好整理成统一结构。
4. CancelledError 不是普通业务失败，捕获后不要吞掉，清理后要继续 raise。

本章仍然使用 asyncio.sleep() 模拟外部 I/O。下一章会把模拟等待换成真实 HTTP 请求。

练习

练习 1：在 04_stable_batch_query.py 的 query_student_async() 中，给学生 5 单独增加等待 2 秒的分支。检查学生 5 是否得到 timeout，同时其他学生的结果仍会打印。

练习 2：在 04_stable_batch_query.py 中把 timeout 分别改成 0.1、0.5、2 秒。检查等待 0.2 秒的普通查询在 0.1 秒时是否超时、在另两个阈值下是否成功，并对比总耗时。

练习 3：先运行 04_stable_batch_query.py 中已有的学生 4 异常分支，观察其 error 结果；再运行 03_cancel_rule.py，观察取消信号是否传到外层。对照学生 3 的 timeout，确认三种处理不同。

练习 4：在 03_cancel_rule.py 的 main() 中，紧接 await task 加入 print("外层认为任务正常结束")，保持它在同一个 try 块内。先运行原有的取消传播版本，确认只出现“外层收到任务取消信号”。再把 long_running_query() 中的 raise 改成 pass，观察是否出现“外层认为任务正常结束”，而不再收到取消信号；最后恢复 raise。

练习 5：在 04_stable_batch_query.py 的 main() 中，把 limit=3 分别改成 limit=0 和 limit=-1。确认程序立即报出 ValueError: limit 必须大于等于 1，而不是停在排队阶段；恢复 limit=3 后，确认原有查询结果正常输出。