异步编程详解

1. 学习目标

本章节对标 MIT 6.006 算法导论、Stanford CS246 并发系统、CMU 15-440 计算机网络、UC Berkeley CS162 操作系统等顶级高校课程对并发与异步编程的教学水准，系统讲解 Python asyncio 库的工程化使用与底层理论。完成本章节学习后，读者应能够：

1.1 Bloom 认知层级目标

1.2 知识地图

[理论基础] 并发 vs 并行 | 协作式调度 | 有限状态机 | CSP 模型
    ↓
[Python 实现] async/await | Coroutine | Task | Future | EventLoop
    ↓
[同步原语] Lock | Semaphore | Event | Condition | Queue | Barrier
    ↓
[工程实战] HTTP 并发 | 数据库异步 | 限流 | 超时 | 优雅关闭
    ↓
[高级话题] TaskGroup | Task 取消语义 | 异步生成器 | uvloop | anyio

1.3 前置知识检查

学习本章节前，请确认你已掌握：

1. Python 函数、装饰器、上下文管理器基础；
2. 迭代器协议（__iter__、__next__）与生成器（yield）；
3. 操作系统线程、进程的基本概念；
4. TCP/IP 网络编程基础（socket、I/O 多路复用）；
5. 集合论与状态机的基本概念。

2. 历史动机与发展脉络

异步编程并非 Python 独有概念，其根源可追溯至 1960 年代的操作系统 I/O 调度理论与 1980 年代的事件驱动编程模型。

2.1 异步编程的理论起源

• 1965：Multics 操作系统引入中断驱动 I/O 模型；
• 1969：Unix 引入 select 系统调用（1971 年 V4 版本正式稳定），允许进程同时等待多个文件描述符；
• 1980s：Berkeley Unix 引入 socket 与异步 I/O；
• 1990s：Windows NT 完成 IOCP（I/O Completion Port）模型；
• 2000s：Linux 2.6 引入 epoll，FreeBSD 引入 kqueue，高性能事件循环成熟。

2.2 Python 异步演进时间线

2.3 第三方生态演进

2.4 设计哲学

Python asyncio 的设计原则（PEP 3156）：

1. 显式异步：async/await 让协程边界清晰可见，避免隐式调度；
2. 单线程协作式：协程主动让出控制权，无需锁保护共享状态；
3. 可插拔事件循环：默认循环可替换为 uvloop 或自定义实现；
4. Future 统一抽象：Future 表示异步结果，跨线程、跨进程通用；
5. 与同步代码兼容：通过 asyncio.to_thread() 桥接阻塞调用。

3. 形式化定义

3.1 协程的代数定义

设 $S$ 为程序状态集合，$E$ 为事件集合，$T : S \times E \to S$ 为状态转移函数。协程（coroutine）是一个五元组：

其中：

• $S$ 为可枚举的状态集合；
• $s_0 \in S$ 为初始状态；
• $E$ 为事件集合（包括 await、send、throw、close）；
• $T : S \times E \to S$ 为状态转移函数；
• $Y : S \to V \times S$ 为 yield 函数，返回一个值与新状态。

3.2 协程状态机

Python 协程在执行过程中处于以下状态之一：

状态转移图：

形式化地，协程的状态转移可表示为：

3.3 事件循环的形式化模型

事件循环（Event Loop）可建模为一个无限循环，等待事件并调度对应的回调：

其中：

• $\mathcal{R}$ 为就绪回调队列（ready queue）；
• $\mathcal{F}$ 为等待中的 Future 集合；
• $\mathcal{T}$ 为定时器堆（timer heap，按时间排序）；
• $\text{run}$ 为调度函数，伪代码如下：

def run(loop):
    while not loop.should_stop:
        # 1. 等待 I/O 事件（epoll/kqueue/IOCP）
        io_events = loop.poll_io(timeout=loop.next_timer_interval())
        loop.process_io(io_events)
        # 2. 处理到期定时器
        expired = loop.pop_expired_timers()
        loop.schedule_callbacks(expired)
        # 3. 执行就绪回调
        while loop.ready_queue:
            callback = loop.ready_queue.popleft()
            callback()
        # 4. 处理信号
        loop.process_signals()

3.4 Task 的调度语义

Task 是对协程的包装，由事件循环调度执行。其语义可形式化为：

每次事件循环迭代时，调度器从就绪队列取出 Task，调用 step() 推进协程执行：

• 若协程 await 一个未完成的 Future，Task 挂起，等待 Future 完成；
• 若协程 await 另一个协程，Task 嵌套等待子协程完成；
• 若协程 return，Task 完成，结果存入 Future。

3.5 async/await 的语义等价

async def f() 等价于：

@types.coroutine
def f():
    # 协程逻辑
    yield  # 让出控制权给事件循环

await expr 等价于：

result = yield from expr.__await__()

形式化地，await 必须作用于 awaitable 对象：

4. 理论推导与原理解析

4.1 协作式 vs 抢占式调度

操作系统线程采用抢占式调度（preemptive scheduling）：

协程采用协作式调度（cooperative scheduling）：

协作式的优势：

1. 无需锁保护共享状态（同一时刻只有一个协程运行）；
2. 上下文切换开销极小（约 100ns，线程约 1-10μs）；
3. 状态机清晰，便于调试。

协作式的劣势：

1. 一个协程阻塞会导致整个事件循环阻塞；
2. 不能利用多核 CPU；
3. 需要严格的异步纪律（async all the way down）。

4.2 单线程并发的数学分析

设 $n$ 个 I/O 任务，每个任务平均等待时间 $w$，CPU 处理时间 $c$。则：

• 同步串行：总耗时 $T_{\text{sync}} = n \cdot (w + c)$；
• 异步并发：总耗时 $T_{\text{async}} = \max(w_1, w_2, \ldots, w_n) + n \cdot c$（理想情况）；
• 加速比：

当 $w \gg c$（I/O 密集型），$S \approx n$；当 $c \gg w$（CPU 密集型），$S \approx 1$。

4.3 GIL 与异步的关系

Python GIL（Global Interpreter Lock）确保同一时刻只有一个线程执行 Python 字节码：

异步代码在单线程内运行，不受 GIL 限制：

但若异步代码调用 CPU 密集型函数（无 await），GIL 仍会阻塞事件循环。

4.4 回调地狱与 async/await 的关系

回调风格的异步代码：

fetchData(url, function (err, data) {
    if (err) return handleError(err);
    processData(data, function (err, result) {
        if (err) return handleError(err);
        saveData(result, function (err) {
            if (err) return handleError(err);
            console.log("Done");
        });
    });
});

async/await 将其线性化：

async def pipeline():
    data = await fetch_data(url)
    result = await process_data(data)
    await save_data(result)
    print("Done")

形式化地，async/await 通过 CPS（Continuation-Passing Style）变换 将回调链转换为状态机，编译器（CPython 解释器）自动生成状态转移代码。

4.5 Future 的代数性质

Future 可视为单子（Monad）：

• return/pure：asyncio.Future() 创建未完成 Future；
• bind/flatMap：await future 将 Future 的值绑定到变量。

满足单子三定律：

1. 左单位律：await Future(value) == value；
2. 右单位律：f = lambda x: Future(x); await f(value) == value；
3. 结合律：await (await future).then(f) == await future.then(g).then(f)。

这使得 Python 异步代码可以组合，如同同步代码一样。

5. 代码示例

5.1 第一个异步程序

"""第一个异步程序。Python 3.12+。"""
from __future__ import annotations

import asyncio


async def say_hello(name: str, delay: float) -> None:
    """异步函数：等待 delay 秒后打印问候。

    Args:
        name: 问候对象名称。
        delay: 等待时长（秒）。
    """
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")


async def main() -> None:
    """主协程：并发执行两个问候任务。"""
    # 并发执行，总耗时约 2 秒（最大 delay）
    await asyncio.gather(
        say_hello("Alice", 2.0),
        say_hello("Bob", 1.0),
    )
    # 输出顺序：Bob（1秒后）, Alice（2秒后）


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

5.2 Task 与 Future

"""Task 与 Future 示例。Python 3.12+。"""
from __future__ import annotations

import asyncio
import time


async def fetch_data(name: str, duration: float) -> str:
    """模拟异步数据获取。"""
    print(f"[{time.strftime('%H:%M:%S')}] {name} 开始获取")
    await asyncio.sleep(duration)
    print(f"[{time.strftime('%H:%M:%S')}] {name} 完成")
    return f"{name}_result"


async def main() -> None:
    # 方式一：使用 create_task 创建并发任务
    task1 = asyncio.create_task(fetch_data("API-1", 2.0), name="fetch-api-1")
    task2 = asyncio.create_task(fetch_data("API-2", 1.5), name="fetch-api-2")

    # 等待单个任务完成（不取消其他任务）
    done, pending = await asyncio.wait(
        {task1, task2},
        return_when=asyncio.FIRST_COMPLETED,
    )
    print(f"首个完成的任务: {done.pop().get_name()}")

    # 等待剩余任务
    await asyncio.wait(pending)

    # 方式二：使用 gather 收集所有结果（保持顺序）
    results = await asyncio.gather(
        fetch_data("API-3", 0.5),
        fetch_data("API-4", 0.3),
        return_exceptions=True,  # 异常作为结果返回，不中断其他任务
    )
    print(f"结果: {results}")

    # 方式三：使用 as_completed 按完成顺序处理
    tasks = [fetch_data(f"API-{i}", 1.0 - i * 0.1) for i in range(5)]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"完成: {result}")


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

5.3 同步原语

"""asyncio 同步原语示例。Python 3.12+。"""
from __future__ import annotations

import asyncio


# 1. Lock：保护共享资源
class AsyncCounter:
    def __init__(self) -> None:
        self._value = 0
        self._lock = asyncio.Lock()

    async def increment(self) -> int:
        async with self._lock:
            self._value += 1
            return self._value

    @property
    def value(self) -> int:
        return self._value


# 2. Semaphore：限制并发数
class ConcurrentLimiter:
    def __init__(self, limit: int) -> None:
        self._sem = asyncio.Semaphore(limit)

    async def run(self, coro_factory):
        async with self._sem:
            return await coro_factory()


# 3. Event：跨协程通知
async def worker(event: asyncio.Event, wid: int) -> None:
    print(f"Worker {wid} 等待事件...")
    await event.wait()
    print(f"Worker {wid} 收到事件，开始工作")


async def controller():
    event = asyncio.Event()
    # 启动 3 个 worker
    workers = [asyncio.create_task(worker(event, i)) for i in range(3)]
    await asyncio.sleep(1.0)
    print("触发事件")
    event.set()  # 唤醒所有等待的 worker
    await asyncio.gather(*workers)


# 4. Condition：复杂条件等待
class AsyncQueue:
    def __init__(self):
        self._items: list = []
        self._cond = asyncio.Condition()

    async def put(self, item):
        async with self._cond:
            self._items.append(item)
            self._cond.notify()

    async def get(self):
        async with self._cond:
            while not self._items:
                await self._cond.wait()
            return self._items.pop(0)


# 5. Barrier：等待多个协程同步
async def phase_worker(barrier: asyncio.Barrier, wid: int):
    print(f"Worker {wid} 阶段 1 开始")
    await asyncio.sleep(0.5 * wid)
    print(f"Worker {wid} 阶段 1 完成，等待其他")
    await barrier.wait()  # 等待所有 worker 完成阶段 1
    print(f"Worker {wid} 阶段 2 开始")


async def main():
    # 演示 Lock
    counter = AsyncCounter()
    await asyncio.gather(*[counter.increment() for _ in range(100)])
    print(f"Counter: {counter.value}")  # 100

    # 演示 Event
    await controller()

    # 演示 Barrier
    barrier = asyncio.Barrier(3)
    await asyncio.gather(*[phase_worker(barrier, i) for i in range(3)])


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

5.4 超时控制

"""超时控制示例。Python 3.11+。"""
from __future__ import annotations

import asyncio


async def slow_operation():
    print("开始慢操作...")
    await asyncio.sleep(10)
    return "完成"


async def main():
    # 方式一：wait_for（兼容旧版本）
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"结果: {result}")
    except asyncio.TimeoutError:
        print("wait_for 超时")

    # 方式二：timeout 上下文管理器（Python 3.11+，推荐）
    try:
        async with asyncio.timeout(2.0):
            result = await slow_operation()
            print(f"结果: {result}")
    except TimeoutError:  # Python 3.11+ 内置 TimeoutError
        print("timeout 上下文超时")

    # 方式三：手动取消 Task
    task = asyncio.create_task(slow_operation())
    await asyncio.sleep(2.0)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task 已取消")

    # 方式四：timeout_at 绝对时间
    deadline = asyncio.get_running_loop().time() + 1.0
    try:
        async with asyncio.timeout_at(deadline):
            await slow_operation()
    except TimeoutError:
        print("timeout_at 超时")


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

5.5 异步迭代器与生成器

"""异步迭代器与生成器示例。Python 3.12+。"""
from __future__ import annotations

import asyncio
from collections.abc import AsyncIterator


# 1. 异步迭代器
class AsyncRange:
    """异步范围迭代器。"""

    def __init__(self, count: int, delay: float = 0.1) -> None:
        self.count = count
        self.delay = delay

    def __aiter__(self) -> AsyncIterator[int]:
        self._i = 0
        return self

    async def __anext__(self) -> int:
        if self._i >= self.count:
            raise StopAsyncIteration
        await asyncio.sleep(self.delay)
        self._i += 1
        return self._i


# 2. 异步生成器
async def async_count(start: int = 0, step: int = 1) -> AsyncIterator[int]:
    """无限异步计数器。"""
    i = start
    while True:
        await asyncio.sleep(0.1)
        yield i
        i += step


# 3. 异步文件读取（需要 aiofiles）
async def async_read_lines(filepath: str) -> AsyncIterator[str]:
    """异步逐行读取文件。

    Args:
        filepath: 文件路径。

    Yields:
        每行内容（已去除换行符）。
    """
    try:
        import aiofiles
    except ImportError:
        # 退化方案：使用 asyncio.to_thread 包装同步文件读取
        with open(filepath, "r", encoding="utf-8") as f:
            for line in f:
                await asyncio.sleep(0)  # 让出控制权
                yield line.rstrip()
        return

    async with aiofiles.open(filepath, "r", encoding="utf-8") as f:
        async for line in f:
            yield line.rstrip()


# 4. 异步生成器管道
async def filter_async(source: AsyncIterator[int], predicate) -> AsyncIterator[int]:
    async for item in source:
        if predicate(item):
            yield item


async def map_async(source: AsyncIterator[int], func) -> AsyncIterator:
    async for item in source:
        yield func(item)


async def main():
    # 异步迭代器
    async for num in AsyncRange(5, delay=0.1):
        print(f"AsyncRange: {num}")

    # 异步生成器（取前 5 个）
    async for num in async_count():
        print(f"async_count: {num}")
        if num >= 5:
            break

    # 异步管道：range -> filter -> map
    async def gen():
        for i in range(10):
            await asyncio.sleep(0)
            yield i

    pipeline = map_async(filter_async(gen(), lambda x: x % 2 == 0), lambda x: x * x)
    async for result in pipeline:
        print(f"Pipeline: {result}")


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

5.6 异步队列与生产者-消费者

"""异步队列与生产者-消费者模式。Python 3.12+。"""
from __future__ import annotations

import asyncio
import random


async def producer(queue: asyncio.Queue[int], pid: int, count: int) -> None:
    """生产者：向队列中放入数据。"""
    for i in range(count):
        item = pid * 100 + i
        await queue.put(item)
        print(f"[P{pid}] 生产 {item}")
        await asyncio.sleep(random.uniform(0.05, 0.2))
    # 放入结束标记
    await queue.put(-1)


async def consumer(queue: asyncio.Queue[int], cid: int) -> None:
    """消费者：从队列中取出数据处理。"""
    while True:
        item = await queue.get()
        if item == -1:
            queue.task_done()
            break
        print(f"[C{cid}] 消费 {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
        queue.task_done()


async def main():
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=10)

    # 启动 2 个生产者，3 个消费者
    producers = [producer(queue, pid, 5) for pid in range(2)]
    consumers = [consumer(queue, cid) for cid in range(3)]

    # 等待所有生产者完成
    await asyncio.gather(*producers)

    # 为每个消费者发送一个结束标记
    for _ in range(3):
        await queue.put(-1)

    # 等待所有消费者完成
    await asyncio.gather(*consumers)

    # 确保所有任务完成
    await queue.join()
    print("所有任务处理完成")


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

5.7 并发 HTTP 请求

"""并发 HTTP 请求示例。Python 3.12+。

需要安装：pip install httpx
"""
from __future__ import annotations

import asyncio
import time
from typing import Any

try:
    import httpx
except ImportError:
    httpx = None  # type: ignore


async def fetch_url(client: "httpx.AsyncClient", url: str) -> dict[str, Any]:
    """异步获取 URL 内容。

    Args:
        client: httpx 异步客户端。
        url: 目标 URL。

    Returns:
        包含状态码、内容长度、耗时的字典。
    """
    start = time.perf_counter()
    try:
        response = await client.get(url, timeout=10.0)
        elapsed = time.perf_counter() - start
        return {
            "url": url,
            "status": response.status_code,
            "length": len(response.content),
            "elapsed": elapsed,
        }
    except Exception as e:
        return {
            "url": url,
            "error": str(e),
            "elapsed": time.perf_counter() - start,
        }


async def fetch_all(urls: list[str], concurrency: int = 10) -> list[dict[str, Any]]:
    """并发获取多个 URL，限制并发数。

    Args:
        urls: URL 列表。
        concurrency: 最大并发数。

    Returns:
        每个 URL 的获取结果。
    """
    if httpx is None:
        raise ImportError("请安装 httpx: pip install httpx")

    semaphore = asyncio.Semaphore(concurrency)

    async def limited_fetch(client: "httpx.AsyncClient", url: str):
        async with semaphore:
            return await fetch_url(client, url)

    async with httpx.AsyncClient() as client:
        tasks = [limited_fetch(client, url) for url in urls]
        return await asyncio.gather(*tasks)


async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/get",
        "https://httpbin.org/headers",
    ] * 4  # 20 个请求

    start = time.perf_counter()
    results = await fetch_all(urls, concurrency=5)
    elapsed = time.perf_counter() - start

    success = sum(1 for r in results if "error" not in r)
    print(f"完成 {len(results)} 个请求，成功 {success} 个，耗时 {elapsed:.2f}s")


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

5.8 异步数据库访问

"""异步数据库访问示例（PostgreSQL）。Python 3.12+。

需要安装：pip install asyncpg
"""
from __future__ import annotations

import asyncio
from typing import Any

try:
    import asyncpg
except ImportError:
    asyncpg = None  # type: ignore


class Database:
    """异步数据库连接池封装。"""

    def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self._pool: "asyncpg.Pool | None" = None

    async def connect(self) -> None:
        """初始化连接池。"""
        if asyncpg is None:
            raise ImportError("请安装 asyncpg: pip install asyncpg")
        self._pool = await asyncpg.create_pool(
            self.dsn,
            min_size=self.min_size,
            max_size=self.max_size,
        )

    async def close(self) -> None:
        """关闭连接池。"""
        if self._pool:
            await self._pool.close()

    async def fetch_users(self, limit: int = 100) -> list[dict[str, Any]]:
        """查询用户列表。"""
        async with self._pool.acquire() as conn:
            rows = await conn.fetch(
                "SELECT id, name, email FROM users ORDER BY id LIMIT $1",
                limit,
            )
            return [dict(row) for row in rows]

    async def fetch_user_by_id(self, user_id: int) -> dict[str, Any] | None:
        """根据 ID 查询用户。"""
        async with self._pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT id, name, email FROM users WHERE id = $1",
                user_id,
            )
            return dict(row) if row else None


async def main():
    db = Database("postgresql://user:pass@localhost/mydb")
    await db.connect()
    try:
        # 并发查询
        users_task = asyncio.create_task(db.fetch_users(limit=50))
        user_task = asyncio.create_task(db.fetch_user_by_id(42))
        users, user = await asyncio.gather(users_task, user_task)
        print(f"查询到 {len(users)} 个用户")
        print(f"用户 42: {user}")
    finally:
        await db.close()


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

5.9 优雅关闭与信号处理

"""优雅关闭与信号处理示例。Python 3.12+。"""
from __future__ import annotations

import asyncio
import signal
import sys
from contextlib import asynccontextmanager


class GracefulServer:
    """支持优雅关闭的异步服务器。"""

    def __init__(self):
        self._shutdown_event = asyncio.Event()
        self._tasks: set[asyncio.Task] = set()

    def spawn_task(self, coro) -> asyncio.Task:
        """创建任务并跟踪。"""
        task = asyncio.create_task(coro)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
        return task

    async def worker(self, wid: int):
        """工作协程。"""
        try:
            while not self._shutdown_event.is_set():
                print(f"Worker {wid} 处理任务...")
                await asyncio.sleep(1.0)
        except asyncio.CancelledError:
            print(f"Worker {wid} 收到取消信号，清理资源...")
            await asyncio.sleep(0.5)  # 模拟清理
            print(f"Worker {wid} 已退出")
            raise  # 重新抛出以传播取消

    async def run(self):
        """启动服务器。"""
        loop = asyncio.get_running_loop()

        # 注册信号处理（仅 Unix）
        if sys.platform != "win32":
            for sig in (signal.SIGINT, signal.SIGTERM):
                loop.add_signal_handler(sig, self._shutdown_event.set)

        # 启动 worker
        for i in range(3):
            self.spawn_task(self.worker(i))

        print("服务器启动，按 Ctrl+C 优雅关闭")
        await self._shutdown_event.wait()
        print("收到关闭信号，开始清理...")

        # 取消所有任务
        for task in self._tasks:
            task.cancel()

        # 等待所有任务完成清理
        if self._tasks:
            await asyncio.gather(*self._tasks, return_exceptions=True)
        print("服务器已关闭")


@asynccontextmanager
async def lifespan():
    """FastAPI 风格的 lifespan 上下文管理器。"""
    print("应用启动")
    yield
    print("应用关闭")


async def main():
    server = GracefulServer()
    await server.run()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n强制退出")

5.10 asyncio 与多进程结合

"""asyncio 与多进程结合处理 CPU 密集型任务。Python 3.12+。"""
from __future__ import annotations

import asyncio
import math
from concurrent.futures import ProcessPoolExecutor
from typing import Iterable


def is_prime(n: int) -> bool:
    """判断素数（CPU 密集型）。"""
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True


def count_primes(start: int, end: int) -> int:
    """统计区间内的素数数量。"""
    return sum(1 for i in range(start, end) if is_prime(i))


async def main():
    loop = asyncio.get_running_loop()

    # 区间划分
    ranges = [(i * 100000, (i + 1) * 100000) for i in range(8)]

    # 使用进程池并发计算
    with ProcessPoolExecutor(max_workers=4) as pool:
        tasks = [
            loop.run_in_executor(pool, count_primes, start, end)
            for start, end in ranges
        ]
        results = await asyncio.gather(*tasks)

    total = sum(results)
    print(f"0-800000 内的素数数量: {total}")


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

6. 对比分析

6.1 Python asyncio vs Go goroutine

6.2 Python asyncio vs JavaScript async/await

6.3 Python asyncio vs Rust async

6.4 Python asyncio vs Java CompletableFuture

6.5 异步 vs 多线程 vs 多进程

7. 常见陷阱与最佳实践

7.1 陷阱：忘记 await

# 错误：忘记 await，协程不会执行
async def bad():
    asyncio.sleep(1)  # 警告：Coroutine was never awaited

# 正确
async def good():
    await asyncio.sleep(1)

最佳实践：启用 asyncio 的调试模式，未 await 的协程会触发警告：

import asyncio
import logging

logging.basicConfig(level=logging.DEBUG)
asyncio.get_event_loop().set_debug(True)

7.2 陷阱：在异步代码中调用阻塞函数

# 错误：time.sleep 会阻塞整个事件循环
import time

async def bad():
    time.sleep(1)  # 整个事件循环卡死 1 秒

# 正确
import asyncio

async def good():
    await asyncio.sleep(1)  # 让出控制权

# 如果必须调用阻塞函数，使用 to_thread
async def good_with_blocking():
    await asyncio.to_thread(time.sleep, 1)

最佳实践：使用 asyncio.to_thread()（Python 3.9+）或 loop.run_in_executor() 包装阻塞调用。

7.3 陷阱：嵌套 asyncio.run()

# 错误：嵌套调用 asyncio.run
async def bad():
    asyncio.run(other_coro())  # RuntimeError: asyncio.run() cannot be called from a running event loop

# 正确：直接 await
async def good():
    await other_coro()

最佳实践：asyncio.run() 只在程序入口调用一次。

7.4 陷阱：Task 未被引用被垃圾回收

# 错误：Task 可能被垃圾回收
async def bad():
    asyncio.create_task(background_work())  # 警告：coroutine was never awaited
    # 如果事件循环没有立即调度，Task 可能被 GC

# 正确：保存 Task 引用
async def good():
    task = asyncio.create_task(background_work())
    # ... 其他逻辑
    await task  # 或显式等待

Python 3.11+ 提供了 asyncio.TaskGroup 更安全的管理方式：

async def better():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(background_work1())
        tg.create_task(background_work2())
    # 退出 with 块时自动等待所有 Task 完成

7.5 陷阱：异常被吞没

# 错误：gather 默认会延迟异常到所有任务完成
async def bad():
    await asyncio.gather(
        fail_quickly(),  # 立即抛异常
        slow_operation(),  # 仍在运行
    )  # 抛异常时 slow_operation 已经浪费了时间

# 正确：使用 return_exceptions=True 显式处理
async def good():
    results = await asyncio.gather(
        fail_quickly(),
        slow_operation(),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"任务失败: {r}")

7.6 陷阱：取消传播不正确

# 错误：吞掉 CancelledError
async def bad():
    try:
        await asyncio.sleep(10)
    except:  # 捕获了所有异常，包括 CancelledError
        print("忽略取消")

# 正确：重新抛出 CancelledError
async def good():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("清理资源...")
        raise  # 必须 re-raise
    except Exception as e:
        print(f"其他异常: {e}")

7.7 陷阱：在多线程中使用 asyncio

# 错误：跨线程操作事件循环
import threading

def bad():
    loop = asyncio.get_event_loop()  # 在子线程中可能获取到错误的循环
    loop.run_until_complete(coro())

# 正确：在新线程中创建独立循环
def good():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(coro())
    finally:
        loop.close()

7.8 陷阱：AsyncIterator 未实现 aiter

# 错误：缺少 __aiter__
class BadAsyncIter:
    async def __anext__(self):
        return 1

# 正确
class GoodAsyncIter:
    def __aiter__(self):
        return self

    async def __anext__(self):
        return 1

7.9 最佳实践总结

1. 全链路异步：异步代码必须从入口到底层全部异步，不能中间穿插同步阻塞调用；
2. 使用 TaskGroup（Python 3.11+）替代裸 create_task，获得结构化并发；
3. 使用 timeout：所有网络请求必须有超时，避免无限等待；
4. 正确处理 CancelledError：清理资源后 re-raise；
5. 使用 Semaphore 限流：避免突发流量打垮下游服务；
6. 使用 anyio 抽象层：跨 asyncio/trio 运行时复用代码；
7. 测试覆盖：使用 pytest-asyncio 编写异步测试；
8. 可观测性：集成 structlog/loguru 异步日志、Prometheus 异步指标。

8. 工程实践

8.1 项目结构

my_async_project/
├── pyproject.toml
├── README.md
├── src/
│   └── myapp/
│       ├── __init__.py
│       ├── main.py           # 入口
│       ├── config.py         # 配置
│       ├── db.py             # 数据库
│       ├── http_client.py    # HTTP 客户端
│       ├── services/         # 业务逻辑
│       │   ├── __init__.py
│       │   └── user_service.py
│       └── routes/           # API 路由
│           ├── __init__.py
│           └── users.py
├── tests/
│   ├── conftest.py
│   └── test_user_service.py
└── Dockerfile

8.2 pyproject.toml 配置

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "myapp"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.110",
    "uvicorn[standard]>=0.27",
    "httpx>=0.27",
    "asyncpg>=0.29",
    "pydantic>=2.6",
    "structlog>=24.1",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "ruff>=0.3",
    "mypy>=1.8",
]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

[tool.ruff]
line-length = 100
target-version = "py312"

8.3 性能优化

"""asyncio 性能优化技巧。Python 3.12+。"""
from __future__ import annotations

import asyncio
import time
from typing import Any


# 1. 使用 uvloop 替代默认事件循环（Linux/macOS）
def setup_uvloop():
    """安装 uvloop 提升性能 2-4 倍。"""
    try:
        import uvloop
        uvloop.install()
        print("uvloop 已启用")
    except ImportError:
        print("uvloop 未安装，使用默认循环")


# 2. 批量化处理
async def fetch_batch(client, urls: list[str], batch_size: int = 50):
    """分批处理，避免一次性提交过多任务。"""
    results = []
    for i in range(0, len(urls), batch_size):
        batch = urls[i : i + batch_size]
        tasks = [client.get(url) for url in batch]
        results.extend(await asyncio.gather(*tasks))
    return results


# 3. 连接复用
class HttpClientPool:
    """HTTP 客户端连接池。"""

    def __init__(self, max_connections: int = 100):
        self._max_connections = max_connections
        self._client = None

    async def __aenter__(self):
        import httpx
        self._client = httpx.AsyncClient(
            limits=httpx.Limits(max_connections=self._max_connections),
            timeout=httpx.Timeout(30.0),
        )
        return self

    async def __aexit__(self, *args):
        if self._client:
            await self._client.aclose()


# 4. 缓存
from functools import lru_cache


class AsyncCache:
    """异步缓存装饰器。"""

    def __init__(self, ttl: float = 300):
        self.ttl = ttl
        self._cache: dict = {}

    def __call__(self, func):
        async def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            if key in self._cache:
                value, expire_at = self._cache[key]
                if time.time() < expire_at:
                    return value
            value = await func(*args, **kwargs)
            self._cache[key] = (value, time.time() + self.ttl)
            return value
        return wrapper


# 5. 背压控制
class BoundedProcessor:
    """有界处理器，防止任务积压。"""

    def __init__(self, max_queue_size: int = 1000):
        self._queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue_size)
        self._workers = 10

    async def submit(self, task):
        """提交任务，队列满时阻塞。"""
        await self._queue.put(task)

    async def worker(self):
        while True:
            task = await self._queue.get()
            try:
                await task()
            finally:
                self._queue.task_done()

    async def start(self):
        for _ in range(self._workers):
            asyncio.create_task(self.worker())

8.4 调试技巧

"""asyncio 调试技巧。"""
import asyncio
import logging
import tracemalloc


def enable_debug():
    """启用 asyncio 调试模式。"""
    logging.basicConfig(level=logging.DEBUG)

    # 1. 启用 asyncio 调试
    asyncio.get_event_loop().set_debug(True)

    # 2. 启用内存分配追踪（检测未 await 的协程）
    tracemalloc.start()

    # 3. 慢回调阈值（默认 100ms）
    asyncio.get_event_loop().slow_callback_duration = 0.1


# 检测阻塞调用
async def detect_blocking():
    """检测阻塞调用。"""
    loop = asyncio.get_running_loop()

    # 使用 debug 模式，阻塞超过 100ms 的回调会被记录
    # 输出类似：Executing <Task ...> took 0.250 seconds
    import time
    start = time.perf_counter()
    await asyncio.sleep(0)
    elapsed = time.perf_counter() - start
    if elapsed > 0.01:
        print(f"警告：事件循环被阻塞 {elapsed * 1000:.1f}ms")


# 使用 aiomonitor 实时监控
async def monitor():
    """使用 aiomonitor 监控事件循环。"""
    try:
        import aiomonitor
        with aiomonitor.start_monitor(loop=asyncio.get_running_loop()):
            # 监控端口默认为 50123，可通过 telnet 连接
            print("监控已启动，连接: telnet localhost 50123")
            await asyncio.sleep(3600)
    except ImportError:
        print("请安装 aiomonitor: pip install aiomonitor")

8.5 测试

"""异步测试示例。Python 3.12+。

需要：pip install pytest pytest-asyncio
"""
import asyncio
import pytest
import pytest_asyncio
from typing import AsyncIterator


@pytest_asyncio.fixture
async def db_connection():
    """异步 fixture。"""
    print("Setup")
    yield {"connected": True}
    print("Teardown")


@pytest.mark.asyncio
async def test_basic_async():
    """基础异步测试。"""
    result = await asyncio.sleep(0.1, result="done")
    assert result == "done"


@pytest.mark.asyncio
async def test_concurrent_operations():
    """测试并发操作。"""
    async def task(n):
        await asyncio.sleep(0.1)
        return n * 2

    results = await asyncio.gather(*[task(i) for i in range(5)])
    assert results == [0, 2, 4, 6, 8]


@pytest.mark.asyncio
async def test_timeout():
    """测试超时。"""
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(asyncio.sleep(10), timeout=0.1)


@pytest.mark.asyncio
async def test_cancel():
    """测试取消。"""
    task = asyncio.create_task(asyncio.sleep(10))
    await asyncio.sleep(0.1)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

8.6 部署

# Dockerfile 示例
FROM python:3.12-slim

WORKDIR /app

# 安装 uvloop 依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc libffi-dev \
    && rm -rf /var/lib/apt/lists/*

# 安装依赖
COPY pyproject.toml .
RUN pip install --no-cache-dir uvloop && pip install --no-cache-dir -e .

# 复制源码
COPY src/ ./src/

# 运行
EXPOSE 8000
CMD ["uvicorn", "myapp.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

9. 案例研究

9.1 Instagram 异步架构

Instagram 早期基于 Django（同步框架），随着用户增长遇到 I/O 瓶颈：

• 问题：图片上传时需调用多个外部服务（S3、CDN、数据库），同步等待导致响应时间累积；
• 方案：使用 asyncio + uvloop 重构上传管线，将串行的外部调用改为并发；
• 结果：上传延迟降低 40%，单机吞吐量提升 3 倍。

关键代码模式：

async def upload_pipeline(image: bytes, user_id: int):
    """Instagram 风格的异步上传管线。"""
    # 并发执行多个 IO 操作
    metadata, thumbnail, watermark = await asyncio.gather(
        extract_metadata(image),
        generate_thumbnail(image),
        apply_watermark(image, user_id),
    )
    # 并发上传到 S3 和 CDN
    await asyncio.gather(
        upload_to_s3(image, key=f"original/{user_id}"),
        upload_to_cdn(thumbnail, key=f"thumb/{user_id}"),
    )
    # 写入数据库
    await db.save_record(user_id, metadata)

9.2 YouTube 视频处理

YouTube 使用 Python + asyncio 处理视频元数据的异步聚合：

• 场景：视频页面需要聚合视频信息、评论、推荐、广告、字幕等多个服务的数据；
• 方案：使用 asyncio.gather 并发请求多个微服务，设置不同超时；
• 优化：对非关键服务（如推荐）使用降级策略，超时返回空列表。

async def get_video_page(video_id: str):
    """YouTube 风格的视频页面聚合。"""
    # 关键服务：短超时
    video_info = await asyncio.wait_for(
        video_service.get(video_id), timeout=0.5
    )

    # 次要服务：中超时
    comments, recommendations = await asyncio.gather(
        comment_service.list(video_id),
        recommendation_service.get(video_id),
        return_exceptions=True,
    )

    # 非关键服务：长超时，失败降级
    try:
        ads = await asyncio.wait_for(
            ad_service.get(video_id), timeout=1.0
        )
    except asyncio.TimeoutError:
        ads = []  # 降级：不显示广告

    return {
        "video": video_info,
        "comments": comments if not isinstance(comments, Exception) else [],
        "recommendations": recommendations if not isinstance(recommendations, Exception) else [],
        "ads": ads,
    }

9.3 Dropbox 文件同步

Dropbox 使用 asyncio 重构桌面客户端：

• 挑战：需要同时处理本地文件系统监控、网络上传、用户界面响应；
• 方案：使用 watchdog（异步文件监控）+ asyncio.Queue + 协程工作池；
• 效果：CPU 占用降低 50%，UI 响应延迟从 200ms 降至 20ms。

import asyncio
from pathlib import Path
from watchfiles import awatch


class FileSync:
    """Dropbox 风格的文件同步客户端。"""

    def __init__(self, root: Path):
        self.root = root
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
        self._workers = 4

    async def watch(self):
        """监控文件变化。"""
        async for changes in awatch(self.root):
            for change_type, path in changes:
                await self.queue.put((change_type, path))

    async def worker(self, wid: int):
        """处理文件变更。"""
        while True:
            change_type, path = await self.queue.get()
            try:
                if change_type.name == "added":
                    await self.upload_file(path)
                elif change_type.name == "modified":
                    await self.sync_file(path)
                elif change_type.name == "deleted":
                    await self.delete_remote(path)
            except Exception as e:
                print(f"Worker {wid} error: {e}")
            finally:
                self.queue.task_done()

    async def upload_file(self, path: str):
        """上传文件到云端。"""
        # 实际实现会使用 httpx 异步上传
        await asyncio.sleep(0.1)
        print(f"Uploaded: {path}")

    async def sync_file(self, path: str):
        """同步文件。"""
        await asyncio.sleep(0.1)
        print(f"Synced: {path}")

    async def delete_remote(self, path: str):
        """删除远程文件。"""
        await asyncio.sleep(0.1)
        print(f"Deleted: {path}")

    async def run(self):
        """启动文件同步服务。"""
        workers = [asyncio.create_task(self.worker(i)) for i in range(self._workers)]
        await self.watch()

9.4 Django 异步视图

Django 3.1+ 引入异步视图，4.0+ 完整支持异步 ORM：

"""Django 异步视图示例。Python 3.12+。"""
from django.http import JsonResponse
from django.views import View
import asyncio
import httpx


async def fetch_user_data(user_id: int):
    """异步获取用户数据。"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()


async def async_user_view(request, user_id: int):
    """Django 异步视图。"""
    data = await fetch_user_data(user_id)
    return JsonResponse({"user": data})


class AsyncDashboardView(View):
    """异步仪表盘视图。"""

    async def get(self, request):
        """并发获取多个数据源。"""
        user_id = request.user.id
        # 并发获取用户、订单、消息
        user, orders, messages = await asyncio.gather(
            self.fetch_user(user_id),
            self.fetch_orders(user_id),
            self.fetch_messages(user_id),
        )
        return JsonResponse({
            "user": user,
            "orders": orders,
            "messages": messages,
        })

    async def fetch_user(self, user_id):
        await asyncio.sleep(0.1)
        return {"id": user_id, "name": "Alice"}

    async def fetch_orders(self, user_id):
        await asyncio.sleep(0.2)
        return [{"id": 1, "total": 99.9}]

    async def fetch_messages(self, user_id):
        await asyncio.sleep(0.15)
        return [{"id": 1, "text": "Hello"}]

9.5 FastAPI 高性能 API

FastAPI 基于 Starlette，原生支持异步：

"""FastAPI 异步 API 示例。Python 3.12+。"""
from __future__ import annotations

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str
    email: str


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理。"""
    # 启动时初始化资源
    print("启动数据库连接池...")
    app.state.db = await init_db_pool()
    yield
    # 关闭时清理资源
    print("关闭数据库连接池...")
    await app.state.db.close()


app = FastAPI(lifespan=lifespan)


async def init_db_pool():
    """初始化数据库连接池。"""
    # 实际使用 asyncpg.create_pool
    return {"connected": True}


@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):
    """获取用户信息。"""
    # 模拟数据库查询
    await asyncio.sleep(0.1)
    if user_id == 1:
        return User(id=1, name="Alice", email="alice@example.com")
    raise HTTPException(status_code=404, detail="User not found")


@app.get("/users")
async def list_users(limit: int = 10):
    """获取用户列表。"""
    # 并发查询多个数据源
    users, total = await asyncio.gather(
        fetch_users_from_db(limit),
        count_users_in_db(),
    )
    return {"users": users, "total": total}


async def fetch_users_from_db(limit: int):
    await asyncio.sleep(0.1)
    return [User(id=i, name=f"User{i}", email=f"user{i}@example.com") for i in range(limit)]


async def count_users_in_db():
    await asyncio.sleep(0.05)
    return 1000


# 启动：uvicorn main:app --reload

10. 练习

10.1 选择题

1. 以下代码的输出是什么？

import asyncio

async def task1():
    print("A")
    await asyncio.sleep(1)
    print("B")

async def task2():
    print("C")
    await asyncio.sleep(0.5)
    print("D")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

A. A B C D B. A C D B C. C A D B D. A C B D

2. 以下哪种情况会导致事件循环阻塞？

A. await asyncio.sleep(1) B. await some_async_function() C. time.sleep(1) D. await asyncio.gather(...)

3. Python 3.11 引入的 asyncio.TaskGroup 的主要优势是什么？

A. 性能比 gather 更快 B. 提供结构化并发，更好的错误处理 C. 支持取消单个任务 D. 可以嵌套使用

4. 关于 asyncio 的 GIL，以下说法正确的是？

A. asyncio 不受 GIL 限制，可以充分利用多核 B. asyncio 在单线程内运行，GIL 不影响其并发能力 C. asyncio 通过多线程绕过 GIL D. GIL 仅影响 CPU 密集型异步代码

5. 以下代码有什么问题？

async def fetch_all(urls):
    tasks = [fetch(url) for url in urls]
    return tasks  # 没有等待

A. 缺少 await B. 返回的是协程对象列表，不是结果 C. 没有创建 Task D. 没有错误

async def fetch_all(urls):
    tasks = [fetch(url) for url in urls]
    return await asyncio.gather(*tasks)

10.2 填空题

1. asyncio.run() 是 Python 3.7+ 的简化 API，等价于以下代码：

loop = asyncio.__________()
try:
    return loop.__________ _________(main())
finally:
    loop.__________()

loop = asyncio.new_event_loop()
try:
    return loop.run_until_complete(main())
finally:
    loop.close()

2. 异步生成器使用 ________ 关键字返回值，异步迭代器需要实现 ________ 和 ________ 方法。

3. Python 3.11 引入的 ________ 上下文管理器替代了 asyncio.wait_for，更符合 Python 风格。

4. 在异步代码中调用同步阻塞函数应使用 ________ 函数（Python 3.9+）。

5. asyncio.gather(*tasks, return_exceptions=________) 可以让异常作为结果返回，不中断其他任务。

10.3 编程题

1. 实现异步限流器

实现一个 RateLimiter 类，限制每秒最多 N 次操作：

class RateLimiter:
    def __init__(self, rate: int):
        """rate: 每秒允许的操作数"""
        self.rate = rate
        # ...

    async def acquire(self):
        """获取一个许可，若超出限制则等待"""
        # ...

import asyncio
import time


class RateLimiter:
    """令牌桶限流器。"""

    def __init__(self, rate: int):
        self.rate = rate  # 每秒令牌数
        self.interval = 1.0 / rate
        self._last_time = 0.0
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            wait_time = self._last_time + self.interval - now
            if wait_time > 0:
                await asyncio.sleep(wait_time)
                self._last_time = time.monotonic()
            else:
                self._last_time = now


# 测试
async def main():
    limiter = RateLimiter(rate=5)  # 每秒 5 次
    for i in range(10):
        await limiter.acquire()
        print(f"[{time.strftime('%H:%M:%S')}] 操作 {i}")


asyncio.run(main())

2. 实现异步缓存装饰器

实现一个带 TTL 的异步缓存装饰器：

def async_cache(ttl: float = 60.0):
    """异步缓存装饰器。"""
    # ...

import asyncio
import time
from collections import OrderedDict
from functools import wraps


def async_cache(ttl: float = 60.0, maxsize: int = 128):
    """异步缓存装饰器，支持 TTL 和 LRU。

    Args:
        ttl: 缓存存活时间（秒）。
        maxsize: 最大缓存项数。
    """
    def decorator(func):
        cache: OrderedDict = OrderedDict()
        lock = asyncio.Lock()

        @wraps(func)
        async def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            async with lock:
                if key in cache:
                    value, expire_at = cache[key]
                    if time.monotonic() < expire_at:
                        cache.move_to_end(key)
                        return value
                    else:
                        del cache[key]

            result = await func(*args, **kwargs)

            async with lock:
                cache[key] = (result, time.monotonic() + ttl)
                if len(cache) > maxsize:
                    cache.popitem(last=False)

            return result
        return wrapper
    return decorator


# 测试
@async_cache(ttl=5)
async def slow_fetch(url: str):
    print(f"Fetching {url}")
    await asyncio.sleep(1)
    return f"result of {url}"


async def main():
    print(await slow_fetch("a"))  # 慢
    print(await slow_fetch("a"))  # 快（缓存）
    await asyncio.sleep(6)
    print(await slow_fetch("a"))  # 慢（缓存过期）


asyncio.run(main())

3. 实现异步任务编排器

实现一个任务编排器，支持依赖关系：

class TaskOrchestrator:
    async def add(self, name: str, coro_factory, depends_on: list[str] = []):
        """添加任务，可指定依赖"""
        # ...

    async def run(self):
        """按依赖顺序执行所有任务"""
        # ...

import asyncio
from typing import Callable, Coroutine, Any


class TaskOrchestrator:
    """异步任务编排器，支持依赖关系。"""

    def __init__(self):
        self._tasks: dict[str, tuple[Callable, list[str]]] = {}
        self._results: dict[str, Any] = {}

    def add(self, name: str, coro_factory: Callable[..., Coroutine], depends_on: list[str] = []):
        """添加任务。

        Args:
            name: 任务名。
            coro_factory: 协程工厂函数，接收依赖结果作为参数。
            depends_on: 依赖的任务名列表。
        """
        self._tasks[name] = (coro_factory, depends_on)

    async def _run_task(self, name: str, completed: dict[str, asyncio.Event]):
        """执行单个任务。"""
        coro_factory, deps = self._tasks[name]
        # 等待所有依赖完成
        for dep in deps:
            await completed[dep].wait()
        # 收集依赖结果
        dep_results = {dep: self._results[dep] for dep in deps}
        # 执行任务
        self._results[name] = await coro_factory(**dep_results)
        completed[name].set()

    async def run(self) -> dict[str, Any]:
        """按依赖顺序执行所有任务。"""
        completed = {name: asyncio.Event() for name in self._tasks}
        tasks = [
            asyncio.create_task(self._run_task(name, completed))
            for name in self._tasks
        ]
        await asyncio.gather(*tasks)
        return self._results


# 测试
async def main():
    orch = TaskOrchestrator()

    async def fetch_user():
        await asyncio.sleep(0.5)
        return {"id": 1, "name": "Alice"}

    async def fetch_orders(user):
        await asyncio.sleep(0.3)
        return [{"id": 1, "user_id": user["id"]}]

    async def fetch_messages(user):
        await asyncio.sleep(0.4)
        return [{"id": 1, "user_id": user["id"]}]

    async def aggregate(user, orders, messages):
        return {
            "user": user,
            "orders": orders,
            "messages": messages,
        }

    orch.add("user", fetch_user)
    orch.add("orders", lambda: fetch_orders(None), depends_on=["user"])  # 简化
    orch.add("messages", lambda: fetch_messages(None), depends_on=["user"])
    orch.add("result", lambda: aggregate(None, None, None), depends_on=["user", "orders", "messages"])

    results = await orch.run()
    print(results)


asyncio.run(main())

10.4 思考题

1. 为什么 asyncio 不能很好地处理 CPU 密集型任务？如何解决？

2. 描述 asyncio.gather 与 asyncio.wait 的区别，并说明各自的适用场景。

# gather：等所有完成，收集结果
results = await asyncio.gather(t1(), t2(), t3())

# wait：第一个完成就返回
done, pending = await asyncio.wait(
    {t1(), t2(), t3()},
    return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
    task.cancel()  # 取消未完成的任务

3. 在生产环境中，如何监控和排查 asyncio 应用的性能问题？

loop = asyncio.get_running_loop()
loop.slow_callback_duration = 0.1  # 记录执行超过 100ms 的回调
loop.set_debug(True)

import tracemalloc
tracemalloc.start()

import aiomonitor
with aiomonitor.start_monitor(loop=loop):
    # telnet localhost 50123
    await app.run()

import structlog
log = structlog.get_logger()

async def handler():
    log.info("request_start", url=url)
    # ...
    log.info("request_end", duration=elapsed)

from opentelemetry.instrumentation.asyncio import AsyncioInstrumentor
AsyncioInstrumentor().instrument()

11. 参考文献

11.1 标准与规范

• [1] Python Software Foundation. Python Language Reference. Python 3.12. https://docs.python.org/3.12/reference/index.html
• [2] PEP 255 - Simple Generators. 2001. https://peps.python.org/pep-0255/
• [3] PEP 342 - Coroutines via Enhanced Generators. 2005. https://peps.python.org/pep-0342/
• [4] PEP 380 - Syntax for Delegating to a Subgenerator. 2011. https://peps.python.org/pep-0380/
• [5] PEP 492 - Coroutines with async and await Syntax. 2015. https://peps.python.org/pep-0492/
• [6] PEP 3156 - Asynchronous IO Support Rebooted. 2012. https://peps.python.org/pep-3156/
• [7] PEP 654 - Exception Groups and except*. 2021. https://peps.python.org/pep-0654/

11.2 学术论文

• [8] Hoare, C. A. R. (1978). Communicating sequential processes. Communications of the ACM, 21(8), 666-677. https://doi.org/10.1145/359576.359585
• [9] Lattner, C., & Adve, V. (2004). LLVM: A compilation framework for lifelong program analysis & transformation. Proceedings of the International Symposium on Code Generation and Optimization, 75-86. https://doi.org/10.1109/CGO.2004.1281665
• [10] Marlow, S., Peyton Jones, S., & Singh, S. (2009). Runtime support for multicore Haskell. Proceedings of the 14th ACM SIGPLAN International Conference on Functional Programming, 65-78. https://doi.org/10.1145/1596550.1596564
• [11] Srinivasan, S., & Mycroft, A. (2009). Kilim: Isolation-typed actors for Java. European Conference on Object-Oriented Programming, 104-128. https://doi.org/10.1007/978-3-642-03013-0_6

11.3 技术文档

• [12] asyncio — Asynchronous I/O. Python 3.12 Documentation. https://docs.python.org/3.12/library/asyncio.html
• [13] uvloop Documentation. MagicStack. https://uvloop.readthedocs.io/
• [14] anyio Documentation. https://anyio.readthedocs.io/
• [15] FastAPI Documentation. https://fastapi.tiangolo.com/
• [16] Starlette Documentation. https://www.starlette.io/

11.4 书籍

• [17] Ramalho, L. (2022). Fluent Python (2nd ed.). O’Reilly Media.
• [18] Beazley, D., & Jones, B. K. (2013). Python Cookbook (3rd ed.). O’Reilly Media.
• [19] Cui, Y. (2023). Architecture Patterns with Python. O’Reilly Media.
• [20] Kleppmann, M. (2017). Designing Data-Intensive Applications. O’Reilly Media.

12. 进一步阅读

12.1 进阶主题

1. asyncio 源码分析：阅读 CPython Lib/asyncio/ 目录源码，理解事件循环、Future、Task 的实现细节；
2. uvloop 性能优化：MagicStack 博客文章 “Making 1 million requests with python asyncio”；
3. 结构化并发：Nathaniel J. Smith 的 “Notes on structured concurrency”；
4. trio 框架：对比 asyncio 与 trio 的设计差异；
5. PEP 703：无 GIL Python 提案及其对 asyncio 的影响。

12.2 相关论文

• “Structured Concurrency” - Martin Sústrik (2016)
• “The C10K Problem” - Dan Kegel (1999)
• “Reactive Streams” - JVM 规范，与 asyncio 互补
• “Actor Model” - Hewitt, Bishop, Steiger (1973)

12.3 实战项目

1. 实现一个简单的 HTTP 服务器：基于 asyncio.start_server，支持路由和中间件；
2. 实现一个 WebSocket 服务器：理解帧解析和长连接；
3. 实现一个分布式任务队列：类似 Celery，但基于 asyncio；
4. 参与 FastAPI 开源贡献：阅读 Starlette 源码，提交 PR；
5. 性能基准测试：对比 asyncio、uvloop、Go、Node.js 在 HTTP 服务的吞吐量。

12.4 在线资源

• Real Python - Async IO in Python：https://realpython.com/async-io-python/
• Stack Overflow - asyncio 标签：https://stackoverflow.com/questions/tagged/asyncio
• Python Asyncio Examples：https://github.com/asyncio-docs/asyncio-examples
• Awesome Asyncio：https://github.com/timofurrer/awesome-asyncio

12.5 视频课程

• Nathaniel J. Smith - Trio: Async I/O for Humans（PyCon 2017）
• Yury Selivanov - async/await in Python 3.5 and Why It Is Awesome（PyCon 2016）
• David Beazley - Python Concurrency From the Ground Up（PyCon 2015）

附录 A：asyncio API 速查表

A.1 核心函数

A.2 同步原语

A.3 异步上下文管理器

A.4 异步迭代器

附录 B：常见错误信息

B.1 RuntimeError: asyncio.run() cannot be called from a running event loop

原因：嵌套调用 asyncio.run()。

解决：直接 await coro()。

B.2 RuntimeWarning: coroutine ‘xxx’ was never awaited

原因：调用 async 函数但未 await。

解决：添加 await。

B.3 RuntimeError: Task attached to a different loop

原因：跨事件循环使用 Task 或 Future。

解决：在同一事件循环内创建和使用。

B.4 RuntimeError: Event loop is closed

原因：在已关闭的事件循环上运行。

解决：使用 asyncio.new_event_loop() 创建新循环。

B.5 KeyboardInterrupt 导致协程泄漏

原因：Ctrl+C 中断时未清理 Task。

解决：

try:
    asyncio.run(main())
except KeyboardInterrupt:
    print("\n收到中断信号")

附录 C：版本兼容性表

附录 D：性能基准

基于 echo TCP 服务器基准测试（1000 并发连接，每秒 1000 请求）：

附录 E：相关标准库模块

• concurrent.futures：线程池和进程池；
• multiprocessing：多进程并行；
• threading：多线程（受 GIL 限制）；
• selectors：I/O 多路复用底层接口；
• signal：信号处理；
• asyncio.subprocess：异步子进程；
• asyncio.staggered：交错执行（3.11+）；
• asyncio.taskgroups：任务组（3.11+）。