第 10 章：异步学生信息服务综合实践

到目前为止，学生信息查询助手已经具备很多单点能力：

1. 能用 FastAPI 提供接口。
2. 能用 httpx.AsyncClient 请求外部服务。
3. 能用 asyncio.Semaphore 限制外部请求并发。
4. 能用 asyncio.timeout() 限制等待时间。
5. 能用模拟数据库池读取学生状态。
6. 能用模拟 Redis 缓存减少重复查询。
7. 能用 asyncio.Queue 处理访问日志、统计和通知这类后台任务。

但是这些能力现在还是分散在不同章节里。

本章要把它们整理成一个完整的小型服务，让你看到一个“项目够用版本”的异步服务大概长什么样。

本章解决 4 个问题：

1. 如何把前面章节的异步能力整理成一个完整服务？
2. 哪些能力应该保留在“项目够用版本”里？
3. 如何设计统一返回结构，方便接口调用和排查问题？
4. 如何在不引入复杂项目分层的情况下，让代码清晰、可运行、可讲解？

本章不再大量增加新知识。

重点是整合和收口。

---

本章学习目标

学完本章后，你应该能做到：

1. 理解综合实战章的目的：整合已有能力，不是再开很多新主题。
2. 能梳理一个日常项目够用的异步服务组成。
3. 能把 HTTP 聚合、模拟数据库、模拟 Redis、后台 Queue 整合到同一个 FastAPI 应用中。
4. 能在 lifespan 中统一创建和关闭共享资源。
5. 能设计统一接口返回结构，例如 ok、reason、source、data、background_tasks。
6. 能给接口增加简单调试端点，例如 /health、/debug/cache、/debug/tasks、/debug/queue。
7. 能说明这个小项目适合什么场景，不适合什么场景。
8. 能总结异步项目中的关键检查点：并发限制、超时、异常、资源释放、阻塞代码、队列退出。
9. 能根据完整项目代码独立复现和修改案例。

---

完整服务的请求链路

我们要完成一个项目够用的异步学生信息服务。

它对外提供这个核心接口：

GET /students/{student_id}/summary

调用接口时，程序会完成这些事情：

这不是生产级架构。

它是一个教学用小项目，目标是把前面学过的异步能力串起来：

• FastAPI 负责对外提供接口。
• httpx.AsyncClient 负责异步请求外部服务。
• Semaphore 负责限制外部请求并发数量。
• asyncio.timeout() 负责限制等待时间。
• 模拟数据库池负责演示连接池和资源释放。
• 模拟 Redis 负责演示缓存读取和写入。
• asyncio.Queue 负责演示后台任务。
• lifespan 负责创建和关闭共享资源。
• 调试接口负责观察服务运行状态。

本章先写三个关键片段，再给出完整代码。

---

服务构建阶段

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

1. 服务骨架与共享资源

教学项目的功能边界

项目够用版本不是“生产级完整架构”。

它只保留当前阶段真正需要的能力：

1. 接口能跑。
2. 异步 I/O 不阻塞主流程。
3. 外部请求能限制并发。
4. 慢任务有超时保护。
5. 资源在应用启动时创建，关闭时释放。
6. 非关键任务可以放入后台队列。
7. 出错时有统一返回结构。
8. 有简单调试接口辅助观察。

暂时不做：

1. 真实数据库。
2. 真实 Redis。
3. 生产级消息队列。
4. 复杂目录分层。
5. 复杂依赖注入。
6. 认证权限。
7. Docker 部署。
8. 性能压测和观测系统。

需求：先整理项目需要的共享资源。

这些资源不应该每个请求都重新创建。

功能分析：

1. FastAPI 应用启动时创建共享资源。
2. 通过 app.state 保存资源。
3. 服务关闭时释放资源。
4. 保持单文件结构，便于录课和复现。

文件名：01_service_skeleton.py

01_service_skeleton.py

import asyncio
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI


class FakeAsyncDBPool:
    async def close(self):
        print("关闭模拟数据库连接池")


class FakeAsyncRedis:
    async def aclose(self):
        print("关闭模拟 Redis 客户端")


async def background_worker(queue: asyncio.Queue):
    while True:
        task = await queue.get()

        try:
            if task is None:
                print("后台 worker 收到退出信号")
                return

            print("处理后台任务", task)
            await asyncio.sleep(0.2)
        finally:
            queue.task_done()


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = httpx.AsyncClient(trust_env=False)
    semaphore = asyncio.Semaphore(10)
    db_pool = FakeAsyncDBPool()
    redis = FakeAsyncRedis()
    task_queue = asyncio.Queue(maxsize=100)
    worker_task = asyncio.create_task(background_worker(task_queue))

    app.state.client = client
    app.state.semaphore = semaphore
    app.state.db_pool = db_pool
    app.state.redis = redis
    app.state.task_queue = task_queue
    app.state.worker_task = worker_task

    print("应用启动：共享资源已创建")

    yield

    print("应用关闭：准备释放资源")

    await task_queue.put(None)
    await task_queue.join()
    await worker_task

    await client.aclose()
    await redis.aclose()
    await db_pool.close()


app = FastAPI(lifespan=lifespan)


@app.get("/health")
async def health():
    return {"status": "ok"}

如果还没有按课程入口同步依赖，先在本地 _code_async 目录运行：

uv sync

运行命令：

uv run python -m uvicorn 01_service_skeleton:app --reload --app-dir c10_async_student_info_service

访问地址：

http://127.0.0.1:8000/health

运行结果说明：

访问 /health 会返回：

{"status": "ok"}

停止服务时，终端会看到资源关闭和后台 worker 退出的输出。

注意：

1. lifespan 负责统一创建和关闭共享资源。
2. app.state.client 保存共享 HTTP 客户端。
3. app.state.semaphore 保存共享并发限制器。
4. app.state.db_pool 保存模拟数据库池。
5. app.state.redis 保存模拟 Redis 客户端。
6. app.state.task_queue 保存后台任务队列。
7. app.state.worker_task 保存后台 worker 任务引用。

骨架示例暂不发送 HTTP 请求，客户端设置 trust_env=False，避免本机环境代理影响启动；最终案例访问本地学生服务时也沿用这一设置。

为什么不在每个请求里创建这些资源？

因为创建 HTTP 客户端、数据库池、Redis 客户端都不是零成本操作。真实项目里，每个请求都创建一次，既浪费资源，也容易造成连接数过多。

本章选择单文件教学项目，是为了让学员能直接复制运行。真实项目可以再拆分目录。

---

2. 统一接口返回结构

统一响应结构的作用

如果接口一会儿返回学生字典，一会儿返回字符串，一会儿直接抛异常，前端和调用方都很难处理。

所以我们统一返回类似这样的结构：

{
    "ok": True,
    "reason": "success",
    "source": "db+http",
    "data": {...},
    "background_tasks": [...],
}

字段含义：

• ok：这次请求是否成功。
• reason：成功或失败原因。
• source：数据来自哪里，例如 cache、db+http。
• data：真正的业务数据。
• background_tasks：这次请求放入了哪些后台任务。
• message：错误信息，只有需要时出现。

这个结构不复杂，但对排查问题很有帮助。

需求：给接口设计一个统一返回格式。

功能分析：

1. 成功、失败都返回字典。
2. 用 ok 表示是否成功。
3. 用 reason 表示原因。
4. 用 source 表示数据来源。
5. 用 background_tasks 表示已入队的后台任务。

文件名：02_response_shape.py

02_response_shape.py

def build_response(
    *,
    ok: bool,
    reason: str,
    source: str,
    data=None,
    background_tasks=None,
    message: str | None = None,
):
    response = {
        "ok": ok,
        "reason": reason,
        "source": source,
        "data": data,
        "background_tasks": background_tasks or [],
    }

    if message is not None:
        response["message"] = message

    return response


success_response = build_response(
    ok=True,
    reason="success",
    source="db+http",
    data={
        "student_id": 1,
        "profile": {"name": "张三"},
    },
    background_tasks=["access_log", "update_stats", "send_notice"],
)

not_found_response = build_response(
    ok=False,
    reason="not_found",
    source="none",
    data=None,
    background_tasks=[],
    message="没有找到学生",
)

timeout_response = build_response(
    ok=False,
    reason="timeout",
    source="http",
    data=None,
    background_tasks=[],
    message="外部服务请求超时",
)

print(success_response)
print(not_found_response)
print(timeout_response)

运行命令：

uv run python c10_async_student_info_service/02_response_shape.py

运行结果说明：

你会看到三种返回结构：

1. 查询成功。
2. 查询不到学生。
3. 外部服务超时。

注意：

1. ok 只回答成功还是失败。
2. reason 说明为什么成功或失败。
3. source 说明数据来自缓存、数据库和 HTTP，还是没有数据。
4. background_tasks 说明本次请求已经放入哪些后台任务。
5. message 用来补充错误说明，不是每次都必须有。

统一结构的好处是：前端、调用方和排查人员不用猜接口返回了什么类型。

这一节只构造并打印响应字典；background_tasks 字段展示计划记录的任务类型，不会真的创建队列或执行后台任务。完整案例才会入队。

---

3. 主请求流程与后台任务

主请求与后台任务的边界

接口主流程必须等这些任务：

1. 查询缓存。
2. 缓存未命中时查数据库。
3. 请求外部 HTTP 服务。
4. 组合接口返回结果。

这些任务影响用户能不能拿到结果。

可以放后台的任务：

1. 记录访问日志。
2. 更新访问统计。
3. 发送模拟通知。

这些任务不影响当前接口的主要结果，所以可以先入队，再由后台 worker 慢慢处理。

教学项目的适用范围

本章完整案例是单进程教学项目。

它适合：

1. 学习异步项目基本结构。
2. 学习资源复用和释放。
3. 学习缓存、数据库、HTTP、队列的组合方式。
4. 学习日常异步项目的检查点。

它不适合直接当生产项目使用。

真实项目中，你需要把模拟 DB 换成 SQLAlchemy async 或 asyncpg，把模拟 Redis 换成 redis.asyncio，把内存 Queue 换成 Celery、Dramatiq、Redis Stream、RabbitMQ 或 Kafka。

---

需求：把主接口的关键流程串起来。

主接口要做这些事：

1. 先查缓存。
2. 缓存没有命中，再查模拟数据库。
3. 再请求外部 HTTP 服务。
4. 组合结果。
5. 写入缓存。
6. 标出要交给后台的任务类型。
7. 返回统一结构。

下面先用内存对象模拟资源，单独运行一次主流程；下一步再看完整的 FastAPI 项目。

文件名：03_summary_flow.py

03_summary_flow.py

import asyncio


async def get_student_summary_with_cache(client, db_pool, redis, semaphore, student_id: int):
    cache_key = f"student:{student_id}:summary"

    cached = await redis.get(cache_key)
    if cached is not None:
        return {
            "ok": True,
            "reason": "success",
            "source": "cache",
            "data": cached,
        }

    status = await load_student_status_from_db(db_pool, student_id)
    if status is None:
        return {
            "ok": False,
            "reason": "not_found",
            "source": "db",
            "data": None,
        }

    async with asyncio.timeout(2):
        http_summary = await fetch_student_summary_from_http(
            client,
            student_id,
            semaphore,
        )

    if http_summary is None:
        return {
            "ok": False,
            "reason": "not_found",
            "source": "http",
            "data": None,
        }

    data = {
        "student_id": student_id,
        "status": status,
        "profile": http_summary["profile"],
        "courses": http_summary["courses"],
        "points": http_summary["points"],
    }

    await redis.set(cache_key, data, expire=60)

    return {
        "ok": True,
        "reason": "success",
        "source": "db+http",
        "data": data,
    }


async def get_student_summary(student_id: int):
    result = await get_student_summary_with_cache(
        app.state.client,
        app.state.db_pool,
        app.state.redis,
        app.state.semaphore,
        student_id,
    )

    background_tasks = await enqueue_access_tasks(
        app.state.task_queue,
        student_id,
        result["ok"],
    )

    result["background_tasks"] = background_tasks
    return result


# 下面用内存对象模拟外部资源，让这一节可以独立运行。
# 完整案例会改用 FastAPI lifespan 管理这些资源。
class FakeRedis:
    def __init__(self):
        self.data = {}

    async def get(self, key):
        return self.data.get(key)

    async def set(self, key, value, expire):
        self.data[key] = value


async def load_student_status_from_db(db_pool, student_id):
    if student_id != 1:
        return None
    return {"status": "active", "remark": "正常在读"}


async def fetch_student_summary_from_http(client, student_id, semaphore):
    async with semaphore:
        await asyncio.sleep(0.01)
        return {
            "profile": {"id": student_id, "name": "张三"},
            "courses": ["Python 基础", "异步编程"],
            "points": {"points": 95},
        }


async def enqueue_access_tasks(task_queue, student_id, is_success):
    return ["access_log", "update_stats", "send_notice"]


async def main():
    from types import SimpleNamespace

    global app
    app = SimpleNamespace(
        state=SimpleNamespace(
            client=None,
            db_pool=None,
            redis=FakeRedis(),
            semaphore=asyncio.Semaphore(3),
            task_queue=None,
        )
    )

    for student_id in (1, 1, 404):
        result = await get_student_summary(student_id)
        print(student_id, result["reason"], result["source"], result["background_tasks"])


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

在 _code_async 目录运行：

uv run python c10_async_student_info_service/03_summary_flow.py

可能看到类似输出：

1 success db+http ['access_log', 'update_stats', 'send_notice']
1 success cache ['access_log', 'update_stats', 'send_notice']
404 not_found db ['access_log', 'update_stats', 'send_notice']

第一次查询完成数据库与 HTTP 聚合，第二次命中缓存；不存在的学生从数据库阶段返回。末尾的内存模拟对象只用于这一节独立演示，其中 enqueue_access_tasks() 仅返回任务类型列表，没有执行 queue.put()；真正入队和 worker 处理在完整案例中实现。

运行逻辑说明：

1. redis.get() 先查缓存。
2. 如果缓存命中，主流程直接返回缓存数据。
3. 如果缓存没命中，继续查模拟数据库。
4. 数据库查不到，返回 not_found。
5. 数据库查到后，请求外部 HTTP 聚合信息。
6. 请求外部 HTTP 时，仍然使用 Semaphore 限制并发。
7. asyncio.timeout(2) 限制等待 HTTP 并发名额及完成 HTTP 聚合的时间；此前的缓存、数据库查询和此后的入队不在这 2 秒内。
8. 成功后写入缓存，下次请求可以更快。
9. 这个片段在返回前列出访问日志、统计、通知三类计划任务，尚未写入队列。
10. 完整案例会在接口返回前入队，再由 worker 处理；处理结果不影响当前查询的数据结果。

注意：

1. 缓存、数据库和 HTTP 聚合是主请求链路，必须等结果。
2. 访问日志、统计和通知可以后台处理。
3. 本片段的 enqueue_access_tasks() 是占位函数，只返回任务类型；完整案例同名函数才会调用 queue.put()。
4. 真正处理入队任务的是完整案例中的后台 worker。

---

4. 案例：异步学生信息服务

本节给出最终完整代码。

项目结构

_code_async/c10_async_student_info_service/
├── 01_service_skeleton.py
├── 02_response_shape.py
├── 03_summary_flow.py
└── 04_student_info_service/
    ├── main.py
    ├── student_api_server.py

安装依赖

安装命令：

uv sync

完整代码

文件名：_code_async/c10_async_student_info_service/04_student_info_service/main.py

main.py

import asyncio
import time
from contextlib import asynccontextmanager
from typing import Any

import httpx
from fastapi import FastAPI

BASE_URL = "http://127.0.0.1:8001"

student_status_data = {
    1: {"status": "active", "remark": "正常在读"},
    2: {"status": "active", "remark": "正常在读"},
    3: {"status": "frozen", "remark": "账号冻结"},
}

processed_tasks: list[dict[str, Any]] = []


class FakeConnection:
    async def fetch_student_status(self, student_id: int):
        await asyncio.sleep(0.1)
        return student_status_data.get(student_id)


class FakeAsyncDBPool:
    def __init__(self, pool_size: int = 5):
        self._semaphore = asyncio.Semaphore(pool_size)
        self._closed = False

    @asynccontextmanager
    async def acquire(self):
        if self._closed:
            raise RuntimeError("数据库连接池已关闭")

        await self._semaphore.acquire()

        try:
            yield FakeConnection()
        finally:
            self._semaphore.release()

    async def close(self):
        self._closed = True
        print("关闭模拟数据库连接池")


class FakeAsyncRedis:
    def __init__(self):
        self._data: dict[str, dict[str, Any]] = {}
        self._closed = False

    async def get(self, key: str):
        if self._closed:
            raise RuntimeError("Redis 客户端已关闭")

        await asyncio.sleep(0.05)

        item = self._data.get(key)
        if item is None:
            return None

        expire_at = item.get("expire_at")
        if expire_at is not None and expire_at <= time.time():
            self._data.pop(key, None)
            return None

        return item["value"]

    async def set(self, key: str, value, expire: int | None = None):
        if self._closed:
            raise RuntimeError("Redis 客户端已关闭")

        await asyncio.sleep(0.05)

        expire_at = None
        if expire is not None:
            expire_at = time.time() + expire

        self._data[key] = {
            "value": value,
            "expire_at": expire_at,
        }

    async def dump(self):
        await asyncio.sleep(0.01)
        return {
            key: item["value"]
            for key, item in self._data.items()
        }

    async def aclose(self):
        self._closed = True
        print("关闭模拟 Redis 客户端")


async def fetch_json(
    client: httpx.AsyncClient, url: str, semaphore: asyncio.Semaphore
):
    async with semaphore:
        response = await client.get(url)

    if response.status_code == 404:
        return None

    response.raise_for_status()
    return response.json()


async def fetch_student_profile(
    client: httpx.AsyncClient, student_id: int, semaphore: asyncio.Semaphore
):
    return await fetch_json(client, f"/students/{student_id}/profile", semaphore)


async def fetch_student_courses(
    client: httpx.AsyncClient, student_id: int, semaphore: asyncio.Semaphore
):
    return await fetch_json(client, f"/students/{student_id}/courses", semaphore)


async def fetch_student_points(
    client: httpx.AsyncClient, student_id: int, semaphore: asyncio.Semaphore
):
    return await fetch_json(client, f"/students/{student_id}/points", semaphore)


async def fetch_student_summary_from_http(
    client: httpx.AsyncClient,
    student_id: int,
    semaphore: asyncio.Semaphore,
):
    results = await asyncio.gather(
        fetch_student_profile(client, student_id, semaphore),
        fetch_student_courses(client, student_id, semaphore),
        fetch_student_points(client, student_id, semaphore),
        return_exceptions=True,
    )
    for result in results:
        if isinstance(result, BaseException):
            raise result
    profile, courses, points = results

    if profile is None:
        return None

    return {
        "profile": profile,
        "courses": courses or [],
        "points": points or {"points": 0},
    }


async def load_student_status_from_db(
    db_pool: FakeAsyncDBPool,
    student_id: int,
):
    async with db_pool.acquire() as conn:
        return await conn.fetch_student_status(student_id)


def build_response(
    *,
    ok: bool,
    reason: str,
    source: str,
    data=None,
    background_tasks=None,
    message: str | None = None,
):
    response = {
        "ok": ok,
        "reason": reason,
        "source": source,
        "data": data,
        "background_tasks": background_tasks or [],
    }

    if message is not None:
        response["message"] = message

    return response


async def get_student_summary_with_cache(
    client: httpx.AsyncClient,
    db_pool: FakeAsyncDBPool,
    redis: FakeAsyncRedis,
    semaphore: asyncio.Semaphore,
    student_id: int,
):
    cache_key = f"student:{student_id}:summary"

    cached = await redis.get(cache_key)
    if cached is not None:
        return build_response(
            ok=True,
            reason="success",
            source="cache",
            data=cached,
        )

    status = await load_student_status_from_db(db_pool, student_id)
    if status is None:
        return build_response(
            ok=False,
            reason="not_found",
            source="db",
            data=None,
            message="数据库中没有找到学生状态",
        )

    try:
        async with asyncio.timeout(2):
            http_summary = await fetch_student_summary_from_http(
                client,
                student_id,
                semaphore,
            )
    except (TimeoutError, httpx.TimeoutException):
        return build_response(
            ok=False,
            reason="timeout",
            source="http",
            data=None,
            message="外部学生服务请求超时",
        )

    if http_summary is None:
        return build_response(
            ok=False,
            reason="not_found",
            source="http",
            data=None,
            message="外部学生服务没有找到学生",
        )

    data = {
        "student_id": student_id,
        "status": status,
        "profile": http_summary["profile"],
        "courses": http_summary["courses"],
        "points": http_summary["points"],
    }

    await redis.set(cache_key, data, expire=60)

    return build_response(
        ok=True,
        reason="success",
        source="db+http",
        data=data,
    )


async def handle_background_task(task: dict):
    await asyncio.sleep(0.1)

    if task["type"] == "access_log":
        result = f"记录访问日志：student_id={task['student_id']}"
    elif task["type"] == "update_stats":
        result = f"更新访问统计：student_id={task['student_id']}"
    elif task["type"] == "send_notice":
        result = f"发送模拟通知：student_id={task['student_id']}"
    else:
        result = f"未知任务类型：{task['type']}"

    processed_tasks.append({
        "task": task,
        "result": result,
        "handled_at": round(time.time(), 3),
    })


async def background_worker(queue: asyncio.Queue):
    while True:
        task = await queue.get()

        try:
            if task is None:
                print("后台 worker 收到退出信号")
                return

            await handle_background_task(task)
        except Exception as exc:
            processed_tasks.append({
                "task": task,
                "result": "failed",
                "message": str(exc),
                "handled_at": round(time.time(), 3),
            })
        finally:
            queue.task_done()


async def enqueue_access_tasks(
    queue: asyncio.Queue,
    student_id: int,
    request_ok: bool,
):
    tasks = [
        {
            "type": "access_log",
            "student_id": student_id,
            "request_ok": request_ok,
            "created_at": round(time.time(), 3),
        },
        {
            "type": "update_stats",
            "student_id": student_id,
            "request_ok": request_ok,
            "created_at": round(time.time(), 3),
        },
        {
            "type": "send_notice",
            "student_id": student_id,
            "request_ok": request_ok,
            "created_at": round(time.time(), 3),
        },
    ]

    for task in tasks:
        await queue.put(task)

    return [task["type"] for task in tasks]


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = httpx.AsyncClient(
        base_url=BASE_URL, timeout=3.0, trust_env=False
    )
    db_pool = FakeAsyncDBPool(pool_size=5)
    redis = FakeAsyncRedis()
    semaphore = asyncio.Semaphore(10)
    task_queue = asyncio.Queue(maxsize=100)
    worker_task = asyncio.create_task(background_worker(task_queue))

    app.state.client = client
    app.state.db_pool = db_pool
    app.state.redis = redis
    app.state.semaphore = semaphore
    app.state.task_queue = task_queue
    app.state.worker_task = worker_task

    print("应用启动：异步学生信息服务已准备好")

    yield

    print("应用关闭：准备处理剩余后台任务并释放资源")

    await task_queue.put(None)
    await task_queue.join()
    await worker_task

    await client.aclose()
    await redis.aclose()
    await db_pool.close()


app = FastAPI(
    title="项目够用的异步学生信息服务",
    lifespan=lifespan,
)


@app.get("/health")
async def health():
    return {
        "status": "ok",
        "service": "async-student-service",
    }


@app.get("/students/{student_id}/summary")
async def get_student_summary(student_id: int):
    start_time = time.perf_counter()

    try:
        response = await get_student_summary_with_cache(
            app.state.client,
            app.state.db_pool,
            app.state.redis,
            app.state.semaphore,
            student_id,
        )

        background_tasks = await enqueue_access_tasks(
            app.state.task_queue,
            student_id,
            response["ok"],
        )

        response["background_tasks"] = background_tasks
        response["elapsed_ms"] = round(
            (time.perf_counter() - start_time) * 1000,
            2,
        )

        return response

    except asyncio.CancelledError:
        raise
    except httpx.HTTPError as exc:
        return build_response(
            ok=False,
            reason="http_error",
            source="http",
            data=None,
            message=str(exc),
        )
    except Exception as exc:
        return build_response(
            ok=False,
            reason="error",
            source="service",
            data=None,
            message=str(exc),
        )


@app.get("/debug/cache")
async def debug_cache():
    return {
        "cache": await app.state.redis.dump(),
    }


@app.get("/debug/tasks")
async def debug_tasks():
    return {
        "count": len(processed_tasks),
        "items": processed_tasks,
    }


@app.get("/debug/queue")
async def debug_queue():
    queue: asyncio.Queue = app.state.task_queue
    maxsize = queue.maxsize

    return {
        "qsize": queue.qsize(),
        "maxsize": maxsize,
        "remaining_capacity": maxsize - queue.qsize() if maxsize > 0 else None,
    }

配套的学生接口也要保存到同一个案例目录：

student_api_server.py

import asyncio

from fastapi import FastAPI, HTTPException

app = FastAPI()

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

student_courses = {
    1: ["Python 基础", "异步编程"],
    2: ["Web 后端", "FastAPI 入门"],
    3: ["数据库基础", "Redis 入门"],
}

student_points = {
    1: 95,
    2: 88,
    3: 76,
}


def ensure_student_exists(student_id: int):
    if student_id not in student_profiles:
        raise HTTPException(status_code=404, detail="student not found")


@app.get("/students/{student_id}/profile")
async def get_student_profile(student_id: int):
    ensure_student_exists(student_id)
    return student_profiles[student_id]


@app.get("/students/{student_id}/courses")
async def get_student_courses(student_id: int):
    ensure_student_exists(student_id)
    return {"courses": student_courses.get(student_id, [])}


@app.get("/students/{student_id}/points")
async def get_student_points(student_id: int):
    ensure_student_exists(student_id)

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

    return {"points": student_points.get(student_id, 0)}

启动命令

在两个终端分别从本地 _code_async 目录开始。

终端 A，启动学生服务：

uv run python -m uvicorn student_api_server:app --port 8001 --app-dir c10_async_student_info_service/04_student_info_service

终端 B，启动聚合服务：

uv run python -m uvicorn main:app --port 8000 --app-dir c10_async_student_info_service/04_student_info_service

访问路径

健康检查：

http://127.0.0.1:8000/health

查询存在的学生：

http://127.0.0.1:8000/students/1/summary

再次查询同一个学生，观察缓存命中：

http://127.0.0.1:8000/students/1/summary

查询不存在的学生：

http://127.0.0.1:8000/students/404/summary

查看缓存：

http://127.0.0.1:8000/debug/cache

查看后台任务处理记录：

http://127.0.0.1:8000/debug/tasks

查看队列状态：

http://127.0.0.1:8000/debug/queue

验收顺序

建议按这个顺序测试：

1. 访问 /health，确认服务启动。
2. 访问 /students/1/summary，第一次通常来自 db+http。
3. 再次访问 /students/1/summary，通常来自 cache。
4. 访问 /debug/cache，查看缓存内容。
5. 访问 /debug/tasks，查看后台任务是否被处理。
6. 访问 /debug/queue，查看队列积压情况。
7. 访问 /students/404/summary，观察统一失败结构。
8. 访问 /students/3/summary，观察慢积分接口触发的 timeout 结果。
9. 稍等 worker 处理后再次访问 /debug/tasks，核对后台记录；最后正常停止服务，观察资源关闭信息。

运行结果说明

第一次访问：

{
  "ok": true,
  "reason": "success",
  "source": "db+http",
  "data": {
    "student_id": 1,
    "status": {"status": "active", "remark": "正常在读"},
    "profile": {"id": 1, "name": "张三", "age": 18},
    "courses": ["Python 基础", "异步编程"],
    "points": {"points": 95}
  },
  "background_tasks": ["access_log", "update_stats", "send_notice"],
  "elapsed_ms": 300.0
}

第二次访问同一个学生时，source 通常会变成：

"source": "cache"

说明缓存生效了。

elapsed_ms 是示意值，实际耗时会随本机响应和调度而变化；验收重点是 ok、reason、source 与返回数据。

访问不存在的学生时，可能返回：

{
  "ok": false,
  "reason": "not_found",
  "source": "db",
  "data": null,
  "background_tasks": ["access_log", "update_stats", "send_notice"],
  "message": "数据库中没有找到学生状态"
}

后台任务的执行不在主响应链路中，但接口必须先把任务放进队列。队列有空位时可以很快返回；队列满时，await queue.put() 会等待，响应时间也会增加。worker 独立处理访问日志、统计和通知。

本章为了记录所有访问，成功查询和 not_found 等业务失败都会入队；这与第 9 章“只在成功时入队”的演示策略不同。响应中的 background_tasks 只表示已提交的任务类型，不表示 worker 已完成处理。访问 /debug/tasks 时若记录尚未出现，稍等后再查；/debug/queue 只显示查看瞬间的队列状态。

服务关闭与资源释放

当你停止服务时，lifespan 的关闭阶段会执行：

1. 向队列放入 None 退出信号。
2. await task_queue.join() 等待队列中任务处理完。
3. await worker_task 等待后台 worker 退出。
4. 关闭 AsyncClient。
5. 关闭模拟 Redis。
6. 关闭模拟数据库池。

这就是项目够用的资源释放流程。

---

本章小结

本章把整门小课的核心能力整合成了一个完整小项目。

你已经看到一个项目够用的异步学生信息服务应该包含什么：

1. FastAPI 异步接口。
2. 异步 HTTP 聚合。
3. 并发限制。
4. 超时控制。
5. 模拟数据库池。
6. 模拟 Redis 缓存。
7. 后台任务队列。
8. 应用生命周期资源管理。
9. 统一返回结构。
10. 调试接口。

本章最重要的不是代码有多复杂，而是理解这些能力在项目里分别解决什么问题。

• AsyncClient 解决异步 HTTP 请求。
• Semaphore 解决并发数量控制。
• timeout 解决一直等待的问题。
• 连接池解决连接复用和数量限制。
• Redis 缓存减少重复查询。
• Queue 把非关键任务移出主请求链路。
• lifespan 负责资源创建和释放。
• 统一返回结构降低接口调用和排查成本。

课程能力回顾

这门小课到这里就完成了。

现在你已经能独立完成这些事情：

1. 写出最小异步程序。
2. 正确使用 async def、await、asyncio.run()。
3. 使用 gather() 同时等待多个异步任务。
4. 使用 TaskGroup 组织一组相关任务。
5. 使用 Semaphore 限制并发数量。
6. 使用 asyncio.timeout() 处理慢任务。
7. 处理基础异常和 CancelledError。
8. 使用 httpx.AsyncClient 发起异步 HTTP 请求。
9. 在 FastAPI 中编写异步接口。
10. 识别 time.sleep()、requests.get() 这类阻塞代码。
11. 理解异步数据库和 Redis 的资源管理原则。
12. 使用 asyncio.Queue 编写简单后台任务。
13. 把这些能力整合成一个项目够用的小型服务。

练习

练习 1：先完整跑通本章服务，再按顺序访问 /health、/students/1/summary、调试接口，确认整条链路都能对上。

练习 2：故意关闭某一块能力，比如缓存、后台任务或外部 HTTP 聚合，观察整体服务会少掉什么效果。

练习 3：自己再加一个调试字段或一个简单端点，比如返回最近一次聚合来源，练习在现有结构上做小范围扩展。