异步编程示例

演示 asyncio 的核心用法：协程、并发任务编排（gather）、信号量限流、 异步生成器与超时控制。这些能力是后续构建高并发 Agent 服务的基础。 对应教程：第2卷《Python 进阶》第3章 asyncio。

文件路径: examples/volume-02-python\async_demo.py

---

源代码

"""异步编程示例。

演示 ``asyncio`` 的核心用法：协程、并发任务编排（``gather``）、信号量限流、
异步生成器与超时控制。这些能力是后续构建高并发 Agent 服务的基础。

对应教程：第2卷《Python 进阶》第3章 asyncio。
"""

from __future__ import annotations

import asyncio
import time
from collections.abc import AsyncIterator
from dataclasses import dataclass, field


@dataclass
class FetchResult:
    """模拟一次异步获取的结果。

    Attributes:
        name: 任务名。
        duration: 实际耗时（秒）。
        value: 返回值。
    """

    name: str
    duration: float
    value: str


async def fake_fetch(name: str, delay: float, value: str | None = None) -> FetchResult:
    """模拟一个耗时的异步 IO 操作（如调用 LLM API）。

    Args:
        name: 任务名。
        delay: 模拟耗时（秒）。
        value: 返回值，默认等于 ``name``。

    Returns:
        ``FetchResult``。
    """
    start = time.monotonic()
    await asyncio.sleep(delay)
    elapsed = time.monotonic() - start
    return FetchResult(name=name, duration=elapsed, value=value if value is not None else name)


async def gather_all(tasks: list[tuple[str, float, str | None]]) -> list[FetchResult]:
    """并发执行多个获取任务并返回结果列表。

    Args:
        tasks: ``(name, delay, value)`` 三元组列表。

    Returns:
        按入参顺序返回的 ``FetchResult`` 列表。
    """
    coros = [fake_fetch(n, d, v) for n, d, v in tasks]
    return await asyncio.gather(*coros)


async def gather_with_semaphore(
    tasks: list[tuple[str, float, str | None]],
    limit: int,
) -> list[FetchResult]:
    """使用信号量限制并发度的并发执行。

    Args:
        tasks: ``(name, delay, value)`` 三元组列表。
        limit: 最大并发数。

    Returns:
        ``FetchResult`` 列表。
    """
    semaphore = asyncio.Semaphore(limit)

    async def _run(name: str, delay: float, value: str | None) -> FetchResult:
        async with semaphore:
            return await fake_fetch(name, delay, value)

    return await asyncio.gather(*[_run(n, d, v) for n, d, v in tasks])


async def stream_items(count: int, delay: float = 0.01) -> AsyncIterator[str]:
    """异步生成器：逐项产出，模拟流式响应。

    Args:
        count: 产出数量。
        delay: 每项间隔（秒）。

    Yields:
        形如 ``item-0`` 的字符串。
    """
    for i in range(count):
        await asyncio.sleep(delay)
        yield f"item-{i}"


async def collect_stream(count: int, delay: float = 0.01) -> list[str]:
    """收集 ``stream_items`` 的全部产出。

    Args:
        count: 产出数量。
        delay: 每项间隔。

    Returns:
        字符串列表。
    """
    return [item async for item in stream_items(count, delay)]


async def fetch_with_timeout(name: str, delay: float, timeout: float) -> FetchResult | None:
    """带超时的获取，超时返回 ``None``。

    Args:
        name: 任务名。
        delay: 模拟耗时。
        timeout: 超时阈值。

    Returns:
        ``FetchResult`` 或 ``None``（超时）。
    """
    try:
        return await asyncio.wait_for(fake_fetch(name, delay), timeout=timeout)
    except asyncio.TimeoutError:
        return None


@dataclass
class AsyncCounter:
    """简单的异步计数器，演示 ``asyncio.Lock`` 保护共享状态。

    Attributes:
        value: 当前计数值。
    """

    value: int = 0
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)

    async def increment(self, n: int = 1) -> int:
        """原子地自增。

        Args:
            n: 增量。

        Returns:
            自增后的值。
        """
        async with self._lock:
            self.value += n
            return self.value


async def main() -> None:
    """演示 asyncio 各类用法。"""
    print("== 并发 gather ==")
    results = await gather_all([("a", 0.05, "A"), ("b", 0.02, "B"), ("c", 0.03, "C")])
    for r in results:
        print(f"  {r.name} -> {r.value} ({r.duration:.3f}s)")

    print("\n== 限流并发 (limit=2) ==")
    tasks = [(f"t{i}", 0.05, None) for i in range(5)]
    start = time.monotonic()
    limited = await gather_with_semaphore(tasks, limit=2)
    print(f"  完成 {len(limited)} 个，总耗时 {time.monotonic() - start:.3f}s")

    print("\n== 流式产出 ==")
    items = await collect_stream(3, delay=0.01)
    print(f"  {items}")

    print("\n== 超时控制 ==")
    ok = await fetch_with_timeout("fast", 0.01, timeout=0.5)
    slow = await fetch_with_timeout("slow", 0.5, timeout=0.05)
    print(f"  fast: {ok.value if ok else None}")
    print(f"  slow: {slow}")


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