ai

1. 异步编程是什么？ #

异步指程序在等待 I/O（网络、磁盘、数据库）时不阻塞当前执行流，而是先去做别的任务，等 I/O 完成后再回来继续。
单线程里也能同时「挂着」很多请求，因此特别适合高并发、大量外部调用的 API 服务。
Python 用 async / await 写协程，用标准库 asyncio 调度。
业务里常见组合是 FastAPI 路由 + httpx / aiohttp 异步 HTTP 客户端。
不适合用 async 的场景是 CPU 密集计算（图像处理、大循环、复杂加密）。
CPU 密集任务应交给多进程或线程池；在 async 里硬算只会占满事件循环，反而拖慢所有请求。

	同步	异步
等待 I/O	线程卡住	让出执行权，先干别的
典型场景	简单脚本、传统 Django 视图	高并发 API、批量调外部接口
写法	`def` + `requests`	`async def` + `await`

2. 定义协程函数 #

在 Python 中，用 async def 声明一个协程函数。
协程函数不是普通同步函数，遇到 await 时可以挂起等待、让出控制权。
调用协程函数会得到协程对象，必须用 await 或交给事件循环执行，否则不会自动跑完。

# 导入 asyncio 异步框架
import asyncio
# 定义一个异步函数 fetch_data，接受 name(字符串) 和 delay(浮点数) 两个参数
async def fetch_data(name: str, delay: float):
    # 打印任务开始的信息
    print(f"{name} 开始")
    # 异步休眠 delay 秒
    await asyncio.sleep(delay)
    # 打印任务完成的信息
    print(f"{name} 完成")
    # 返回参数 name
    return name

3. 等待异步操作完成 #

await 用于挂起当前协程，等待一个异步操作完成。
遇到 await 时，Python 把控制权交回事件循环，去运行其它协程。
异步操作完成后，再从 await 处继续执行当前协程。
常见用法是等待 I/O，例如 await asyncio.sleep(2)（不会阻塞整个进程的主线程逻辑）。

await asyncio.sleep(2)  # 等待 2 秒（不会阻塞主线程）

4. 启动事件循环并运行 #

asyncio 的协程必须放进事件循环（event loop）中才能执行。
asyncio.run(main()) 会自动创建并启动事件循环，运行 main 协程，结束后关闭循环。
脚本入口常用 asyncio.run(main())，业务在 FastAPI 等框架里一般由框架托管事件循环。

# 导入 asyncio 异步框架
import asyncio

# 定义一个异步函数 fetch_data，参数 name 是字符串类型，delay 是浮点数类型
async def fetch_data(name: str, delay: float):
    # 打印任务开始的信息，输出 name 加上“开始”
    print(f"{name} 开始")
    # 异步休眠 delay 秒，不会阻塞主线程
    await asyncio.sleep(delay)
    # 打印任务完成的信息，输出 name 加上“完成”
    print(f"{name} 完成")
    # 返回参数 name
    return name

# 定义主异步函数 main
async def main():
    # 调用异步函数 fetch_data，参数为 "A" 和 2，并等待其完成
    await fetch_data("A", 2)

# 执行主异步函数 main
asyncio.run(main())

5. 多个协程并发 #

asyncio.gather(...) 可并发运行多个协程，并等待它们全部完成。
gather 返回结果列表，顺序与传入协程的顺序一致。
高并发 I/O 场景常用此写法；总耗时约等于最慢的那个任务，而不是各任务耗时之和。
日常批量外调优先用 gather，比逐个 await 更高效。

# 导入 asyncio 异步框架
import asyncio

# 定义一个异步函数 fetch_data，参数 name 是字符串类型，delay 是浮点数类型
async def fetch_data(name: str, delay: float):
    # 打印任务开始的信息，输出 name 加上“开始”
    print(f"{name} 开始")
    # 异步休眠 delay 秒，不会阻塞主线程
    await asyncio.sleep(delay)
    # 打印任务完成的信息，输出 name 加上“完成”
    print(f"{name} 完成")
    # 返回参数 name
    return name

# 定义主异步函数 main
async def main():
    # 使用 asyncio.gather 并发运行两个 fetch_data 协程，等待它们都完成，并将返回结果赋值给 results
    results = await asyncio.gather(
        # 启动 fetch_data 协程，参数为 "A" 和 2
        fetch_data("A", 2),
        # 启动 fetch_data 协程，参数为 "B" 和 1
        fetch_data("B", 1),
    )
    # 打印 results 列表，即 ["A", "B"]
    print(results)  # ['A', 'B']

# 执行主异步函数 main
asyncio.run(main())

6. 提前调度、稍后再 await #

需要提前把任务丢进事件循环、稍后再 await 结果时，用 asyncio.create_task(coro)。
create_task 常见于生产消费、后台监视等需要管理任务生命周期的场景。
与 gather 的区别：gather 通常一次性并发并等待全部完成；create_task 可随时创建、单独 await 或取消。

# 导入 asyncio 模块
import asyncio

# 定义一个异步函数 fetch_data，参数 name 是字符串类型，delay 是浮点数类型
async def fetch_data(name: str, delay: float):
    # 打印任务开始的信息，输出 name 加上“开始”
    print(f"{name} 开始")
    # 异步休眠 delay 秒，不会阻塞主线程
    await asyncio.sleep(delay)
    # 打印任务完成的信息，输出 name 加上“完成”
    print(f"{name} 完成")
    # 返回参数 name
    return name

# 定义主异步函数 main
async def main():
    # 创建一个协程任务，调用 fetch_data，参数为"A"和2
    task = asyncio.create_task(fetch_data("A", 2))
    # 等待任务完成，并将返回值赋给 result
    result = await task
    # 打印返回的结果
    print(result)

# 执行主异步函数 main
asyncio.run(main())

7. 并发请求外部 API #

真实项目里异步最常见的用途是同时请求多个 URL（聚合第三方数据、健康检查、批量拉配置）。
应使用 httpx.AsyncClient（或 aiohttp），在 async with 里复用同一客户端，连接池由库管理。
比每次请求都新建客户端更高效，也避免连接泄漏。
在 async def 里调用同步 requests.get() 是典型错误：会阻塞整个事件循环，其它协程全部卡住。
需要第三方库时优先选带 async 支持的版本；没有则用 run_in_executor 把阻塞调用丢进线程池（了解即可）。

# 导入 asyncio 模块，用于异步 IO
import asyncio
# 导入 httpx 模块，用于发起 HTTP 请求
import httpx

# 定义一个异步函数 fetch_status，用于获取单个 URL 的状态码
async def fetch_status(client: httpx.AsyncClient, url: str) -> int:
    # 使用传入的 AsyncClient 实例异步请求给定 URL，设置超时时间为 10 秒
    resp = await client.get(url, timeout=10.0)
    # 返回响应的状态码
    return resp.status_code

# 定义主异步函数 main
async def main():
    # 定义一个包含待请求 URL 的列表
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
    ]
    # 使用异步上下文管理器创建 AsyncClient 实例，确保复用连接
    async with httpx.AsyncClient() as client:
        # 并发执行对每个 URL 的 fetch_status 调用，收集所有返回值
        codes = await asyncio.gather(*[fetch_status(client, u) for u in urls])
    # 打印所有状态码组成的列表
    print(codes)

# 执行主异步函数 main，开启事件循环
asyncio.run(main())

安装：pip install httpx 或 uv add httpx。

8. 异常与超时 #

线上异步代码必须考虑超时和部分失败：某个 URL 慢或返回 500，不应拖死整批 gather。
httpx 的 timeout 参数限制单次请求最长等待时间。
try/except 捕获 HTTPError 等与同步写法类似。
gather(..., return_exceptions=True) 表示某个协程失败时，结果列表对应位置是异常对象，其余任务照常完成。
对爬虫、批量调第三方接口，可用 asyncio.Semaphore(n) 限制同时进行的请求数，避免占满连接或打挂对方服务。

# 导入asyncio模块，用于异步编程
import asyncio
# 导入httpx模块，用于发送HTTP请求
import httpx

# 定义一个带超时和异常处理功能的异步请求函数
async def safe_fetch(client, url):
    # 尝试执行以下代码块
    try:
        # 使用client发送GET请求，请求超时设为5秒
        r = await client.get(url, timeout=5.0)
        # 检查请求结果是否有异常，若有则抛出异常
        r.raise_for_status()
        # 返回响应内容
        return r.text
    # 捕获httpx.HTTPError异常
    except httpx.HTTPError as e:
        # 返回错误信息字符串
        return f"失败: {e}"

# 定义主异步函数
async def main():
    # 定义一个URL列表
    urls = [
        # 第一个网址
        "https://www.example.com",
        # 第二个网址，可能访问失败
        "https://www.nonexistentwebsite123456789.com",  # 一个可能失败的地址
        # 第三个网址
        "https://httpbin.org/get"
    ]
    # 创建异步HTTP客户端
    async with httpx.AsyncClient() as client:
        # 为每个URL生成一个safe_fetch任务列表
        tasks = [safe_fetch(client, url) for url in urls]
        # 异步并发执行所有任务，失败的任务返回异常对象而不影响其他
        results = await asyncio.gather(*tasks, return_exceptions=True)
        # 遍历所有请求结果
        for i, result in enumerate(results):
            # 打印每个URL返回的结果
            print(f"URL {i}: {result}")

# 运行主异步函数main()
asyncio.run(main())

9. 项目开发要点 #

落地异步项目时，先自检：是否真是 I/O 瓶颈、链路是否全程 async、外调是否有超时与并发上限。
I/O 密集才上 async；算力密集用进程/线程，别用 async 假装加速 CPU。
一条链路到底：async 视图 → async HTTP/DB 客户端，中间不要夹同步阻塞。
批量外调用 gather + httpx.AsyncClient，注意超时与 return_exceptions。
错误与日志：try/except 与同步类似，见异常章节。
调试：先缩小为单个 await 复现，再查是否误用阻塞库。

10. 总结 #

本章核心：在 I/O 等待时让出执行权，用 async/await 表达协程逻辑。
用 asyncio 与异步 HTTP 客户端并发多路请求。
不要在 async 里做阻塞或 CPU 重活（同步 requests、time.sleep、大段计算）。
下表可作速查。

概念	要点
`async` / `await`	定义与等待协程
`asyncio.run`	程序入口
`gather`	并发多个 I/O
异步 HTTP	`httpx` + `async with Client`
禁忌	在 async 里 `sleep` / 同步 requests

导航菜单

1. 异步编程是什么？ #

2. 定义协程函数 #

3. 等待异步操作完成 #

4. 启动事件循环并运行 #

5. 多个协程并发 #

6. 提前调度、稍后再 await #

7. 并发请求外部 API #

8. 异常与超时 #

9. 项目开发要点 #

10. 总结 #