Python 异步编程实战：asyncio 协程、并发控制与常见坑

一、为什么需要异步

很多服务的瓶颈不在 CPU，而在“等待”：等数据库返回、等 HTTP 响应、等文件读写。传统同步模型下，一个请求在等待时会占着一个线程干等，想提高并发只能不断加线程，线程一多切换开销和内存占用就上来了。

asyncio 的思路是：用一个事件循环（event loop）在单线程里调度大量协程，遇到 IO 等待就切去处理别的任务，等结果就绪再切回来。这样少量线程也能扛住大量并发 IO，特别适合爬虫、网关、聊天服务这类 IO 密集场景。

二、核心三件套：协程、事件循环、await

import asyncio

async def fetch(name, sec):
    print(f"{name} start")
    await asyncio.sleep(sec)      # 模拟 IO 等待，不阻塞事件循环
    print(f"{name} done")
    return name

async def main():
    # 并发执行多个协程
    results = await asyncio.gather(
        fetch("A", 2),
        fetch("B", 1),
        fetch("C", 3),
    )
    print(results)

asyncio.run(main())

要点：

• async def 定义的是协程函数，调用它不会立即执行，而是返回协程对象。
• await 只能出现在 async def 里，用于等待一个可等待对象（协程/Future/Task）。
• asyncio.run() 是最外层入口，负责创建并关闭事件循环。

上面的例子总耗时约 3 秒（取最长那个），而不是 2+1+3=6 秒——因为 gather 让三个任务并发跑。

三、Task：真正被调度的单位

await coro 是“串行等待”，而把协程包成 Task 才会被事件循环并发调度：

async def main():
    t1 = asyncio.create_task(fetch("A", 2))
    t2 = asyncio.create_task(fetch("B", 1))
    # 这里可以做别的事
    await t1
    await t2

asyncio.run(main())

create_task 立即把协程排入事件循环开始跑，await 只是等它结束。忘记保存 Task 的引用可能被垃圾回收，导致任务“莫名消失”——建议用变量接住，或用 asyncio.gather。

四、并发控制与超时

无限制地并发可能打爆下游，用 Semaphore 限流、用 timeout 兜底：

import asyncio

sem = asyncio.Semaphore(10)   # 最多 10 个并发

async def bounded(url):
    async with sem:
        await asyncio.sleep(0.5)
        return url

async def main():
    tasks = [bounded(f"u{i}") for i in range(100)]
    try:
        res = await asyncio.wait_for(asyncio.gather(*tasks), timeout=10)
        print(len(res))
    except asyncio.TimeoutError:
        print("超时")

• Semaphore：控制同时进行的数量，避免瞬间打满连接池。
• wait_for：给一组任务加总超时，防止个别慢请求拖死整体。

五、异步中的常见坑

1. 在协程里调用阻塞函数（如 time.sleep、requests.get、同步 DB 驱动）：会卡住整个事件循环，所有协程一起停摆。应改用异步库（aiohttp、asyncpg、aiomysql），或用 await asyncio.to_thread(fn) 把阻塞调用丢到线程池。
2. CPU 密集任务放协程里：协程不解决 CPU 瓶颈，计算密集应交给多进程（ProcessPoolExecutor）或独立 worker。
3. 事件循环里做大量同步计算：即使没有 IO，长计算也会阻塞调度，注意切分或下放。
4. 吞掉异常：Task 里未捕获的异常默认只记日志不抛出，gather(..., return_exceptions=True) 或给 Task 加 add_done_callback 才能发现。

六、一个实用模式：并发抓取 + 限流 + 重试

import asyncio, aiohttp

async def fetch_one(session, url, sem, retries=3):
    async with sem:
        for attempt in range(retries):
            try:
                async with session.get(url, timeout=5) as r:
                    return await r.text()
            except Exception:
                if attempt == retries - 1:
                    return None
                await asyncio.sleep(2 ** attempt)   # 指数退避

async def run(urls):
    sem = asyncio.Semaphore(20)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, u, sem) for u in urls]
        return await asyncio.gather(*tasks)

# asyncio.run(run(urls))

这个模板覆盖了并发、限流、超时、指数退避重试，是 IO 密集采集类任务的常用骨架。

七、async/await 与线程、多进程怎么选

• IO 密集、高并发：asyncio（协程）最优，开销小、可万分并发。
• 已有同步库、改造量大：线程池（asyncio.to_thread / ThreadPoolExecutor）过渡。
• CPU 密集：多进程，协程无济于事。

八、总结

asyncio 的核心是单线程事件循环 + 协程调度：用 async/await 表达“等待”，用 gather/create_task 实现并发，用 Semaphore/timeout 控制节奏，绝不能在协程里跑阻塞调用。理解这几点，就能把 IO 密集型服务的并发能力用很小成本拉起来。