Python 异步编程实战指南

直接回答：asyncio 让 Python 用单线程协程实现高并发 IO：一个线程同时等待成百上千个网络响应，减少线程调度，但协程共享状态仍可能需要 asyncio.Lock。IO 密集程序（API 聚合、爬虫、微服务）的实际吞吐取决于负载、并发上限与依赖，不能统一保证提升倍数。

异步的心智模型

同步代码：烧水时干等，水开再泡茶。异步代码：烧水的间隙去洗杯子——等待的缝隙被利用起来。关键前提：等待必须是"可让出"的（异步 IO），CPU 计算让不出，异步帮不上忙。

第一个异步程序

import asyncio

async def fetch(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)   # 模拟网络等待
    print(f"{name} 完成")
    return name

async def main():
    results = await asyncio.gather(
        fetch("接口A", 2),
        fetch("接口B", 1),
        fetch("接口C", 3),
    )
    print(results)

asyncio.run(main())

三个任务总耗时 ≈ 3 秒（最慢的那个），同步写法要 6 秒。async def 定义协程、await 让出控制权、gather 并发汇合——三件套就是全部核心。

异步生成器与异步 for

流式数据（分页 API、消息流）的优雅抽象：

async def paginate(url):
    page = 1
    while True:
        data = await fetch_page(url, page)
        if not data["items"]:
            break
        for item in data["items"]:
            yield item
        page += 1

async def consume():
    async for item in paginate("https://api.example.com/list"):
        process(item)

async for 让消费方代码和同步版本几乎一样，背后的网络等待全自动让出。

信号量：给并发装上闸门

大量同时请求可能触发限流或造成过载；信号量不限制已创建任务数量，超大输入还需有界队列及固定 worker。信号量限制并发上限：

import httpx

async def bounded_fetch(sem, client, url):
    async with sem:                        # 同时最多 N 个在执行
        return await client.get(url)

async def crawl_all(urls):
    sem = asyncio.Semaphore(20)
    async with httpx.AsyncClient() as client:
        return await asyncio.gather(*(bounded_fetch(sem, client, u) for u in urls))

避坑清单

1. 同步调用混进协程：requests.get、文件读写、time.sleep 都会阻塞事件循环——换 httpx/aiofiles/asyncio.sleep，或 asyncio.to_thread 兜底；
2. 忘记 await：调用协程不 await 什么都不会发生（只返回协程对象）；
3. gather 与 TaskGroup 异常不同：gather 默认传播第一个异常，但不自动取消其他任务；return_exceptions=True 将异常作为结果。TaskGroup（3.11+）在子任务非取消异常时取消其余任务并聚合异常；
4. CPU 密集别用异步：那是多进程/多解释器的领域。

常见问题（FAQ）

Q：什么时候值得上异步？
A：IO 密集且并发高：API 聚合网关、爬虫、消息处理、WebSocket 服务。CRUD 为主的传统应用同步框架足矣，别为时髦付复杂度税。

Q：asyncio 和多线程怎么选？
A：都是 IO 并发方案。asyncio 可在单事件循环内并发等待 IO，但共享状态仍需同步，容量应按任务资源测量，但要求全链路异步库；线程池对存量同步代码友好（asyncio.to_thread 就是桥梁）。

Q：TaskGroup 和 gather 怎么选？
A：3.11+ 优先 TaskGroup：异常传播语义更安全（一个失败取消全组并聚合异常），任务生命周期不出作用域。gather 留给兼容老代码。

官方参考

本文基于官方文档整理，未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全；局部片段不等同于完整生产应用。

• 官方资料 1

延伸阅读

• Python 3.14 新特性详解
• 用 ipdb 调试 Python 代码完全指南