jianfeng huang

Posted on Oct 2

把一段 10 秒的同步循环改成协程之后，究竟发生了什么？

#python #async #concurrency #httpx

TL;DR：改协程不是“让代码变快”，而是把 N 次“等待”叠成一坨。顺序执行的 30 次请求要 10 秒，协程版能在 1 秒左右结束——省下来的是等待，不是计算。但改完之后你必须补上四件事：超时、限流、异常语义、以及“哪一步其实还是串行的”。

上一篇我们把“协程是什么”讲成了人话：一个能中途暂停、把线程让出去、之后从原地继续的函数。这篇不看定义，看一段真实的烂代码改成协程之后，时间线到底怎么变了。

一、起点：一段人人都写过的同步循环

import time
import requests

URLS = [f"https://api.example.com/items/{i}" for i in range(30)]

def fetch_all_sync(urls):
    results = []
    for url in urls:
        resp = requests.get(url, timeout=5)   # 阻塞在这里等网络
        results.append(resp.json())
    return results

start = time.perf_counter()
fetch_all_sync(URLS)
print(f"同步版耗时 {time.perf_counter() - start:.2f}s")

假设每个接口的响应时间是 300ms 左右（下面所有数字都是量级示意，换成你自己的接口会不一样）。30 个串行请求 ≈ 9 秒，其中 99% 的时间你的进程在睡觉：CPU 什么都不干，就等着 socket 上有字节回来。

这就是协程的靶场：不是算得太慢，是等得太多。

二、改写成协程版本

import asyncio, time
import httpx

URLS = [f"https://api.example.com/items/{i}" for i in range(30)]

async def fetch_one(client, url):
    resp = await client.get(url, timeout=5)    # 挂起点：等 IO，把线程让出去
    return resp.json()

async def fetch_all_async(urls):
    async with httpx.AsyncClient() as client:  # 连接池复用，别在循环里新建 client
        tasks = [fetch_one(client, url) for url in urls]
        return await asyncio.gather(*tasks)

start = time.perf_counter()
asyncio.run(fetch_all_async(URLS))
print(f"协程版耗时 {time.perf_counter() - start:.2f}s")

同样的 30 个请求，量级上会落到 0.5～1 秒。为什么能快 10 倍？

因为 asyncio.gather 把 30 个协程一次性推进到各自的挂起点：第一个请求发出去了、还没回，立刻切去发第二个，30 个请求在重叠的时间窗里都在飞。真正花掉的墙上时间，是“最慢的那个请求”加上调度开销，而不是“30 个请求之和”。

时间线大致是这样（→ 表示等，■ 表示占用 CPU 在跑代码）：

同步版   ■→→→■→→→■→→→■→→→ ...       总时长 ≈ 30 × 300ms
协程版   ■■■■■■■■■■■■■■■■■■■■■■■■■■■■  ← 30 次“→”叠在一起
         ■ 发完就切走，中间的 → 全部重叠

注意最后一行：协程版里，真正写代码的 ■ 段一点没变短，甚至因为调度还多了一点点。 省下来的全是 →。

三、所以“并发度”到底由什么决定？

不是你想开多少就有多少。协程的并发度取决于三件事：连接池上限、对端能承受的速率、以及你自己的限流。 30 个请求一次丢出去看着很爽，3000 个就不一样了——大概率你会先把自己的连接池打满，或者被对端限流甚至封 IP。

这就要自己加限流：

async def fetch_all_async(urls, limit=20):
    sem = asyncio.Semaphore(limit)             # 最多 20 个请求同时在飞
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=limit),   # 连接池也要配一致
        timeout=httpx.Timeout(5.0, connect=2.0),
    ) as client:
        async def guarded(url):
            async with sem:                    # 拿不到名额就在这里排队
                return await fetch_one(client, url)
        return await asyncio.gather(*(guarded(u) for u in urls))

Semaphore 要放在 client 外层只是习惯问题，重点是并发度和连接池要一致：信号量给 20、连接池给 100，等于没限；信号量给 100、连接池给 20，多余的协程会卡在拿连接那一步，看起来像“随机变慢”。

四、改完必须补的四件事

1. 超时（必做）
await 是可以在挂起点上永远等下去的。统一给 client 配 httpx.Timeout(5.0, connect=2.0)，再加一层总时限：

try:
    data = await asyncio.wait_for(fetch_one(client, url), timeout=8)
except asyncio.TimeoutError:
    data = None

2. gather 的异常语义（最容易踩）
asyncio.gather(...) 默认第一个异常会立刻向外抛，但其余任务并不会被取消——它们继续在后台跑，异常被丢掉。想两者都要，用 return_exceptions=True 收敛结果：

results = await asyncio.gather(*(guarded(u) for u in URLS), return_exceptions=True)
ok = [r for r in results if not isinstance(r, Exception)]
fail = [r for r in results if isinstance(r, Exception)]

3. 取消
用户断线、上游超时，都会变成在当前挂起点往协程里扔一个 CancelledError。所以千万别写：

try:
    await do_work()
except Exception:      # 顺手把取消也吞了，任务会假装自己正常结束
    pass

CancelledError 在 3.8+ 是 BaseException，一般不会被子句吞掉，但 except Exception 包住 await 的同款写法仍然常见。清理逻辑放 finally，需要放行就 raise。

4. 别在循环里新建 client
httpx.AsyncClient 一建一拆就是一次连接池的创建与销毁，TLS 握手全白做。整个批处理共用一个 client，退出时 aclose()（用 async with 最省心）。

五、改完没变快？回头查这三处

链路里还藏着同步调用：time.sleep、requests、同步文件/数据库驱动。它们不让出线程，会把整个事件循环按在地上。搜一遍，换成 await asyncio.to_thread() 或异步驱动。
await 写成了串行：for u in urls: await fetch_one(u) 只是把同步循环换了个写法，时间一点没省。要并发，得先把协程造出来（列表推导只会创建协程对象，不执行），再交给 gather / TaskGroup。
量太小或对端有缓存：只有 3 个请求时，协程版的调度开销会让它跟同步版打平。别为了“看起来先进”而改。

六、什么时候不要改

CPU 密集：算哈希、压图片、做排序——协程帮不上忙，交给多进程。
依赖只有同步版：老的 ORM / SDK 没有异步接口，先 to_thread() 包一层，别硬把整条链路改成 async。
调用链很短：一两个请求就结束的脚本，可读性比并发度值钱。

收口

把同步循环改成协程，你并没有让代码变快，你只是让 30 次“等待”重叠成了一坨。 所以判断要不要改，只需要问一句：这段代码的大部分时间，是在等别人，还是在算自己的？答案在“等”，协程就是对的工具；答案在“算”，请去看多进程。

DEV Community