Python 异步编程与 限流 实践

最近在写一个批量拉接口数据的小工具，接口本身不算慢，但一上 asyncio.gather 把几百个请求同时甩出去，前几秒很爽，后面就开始 429，偶尔还给你来个连接超时。日志里一堆 Too Many Requests，看着就头大。后来才老老实实把限流加上，发现异步编程里“会并发”和“会控制并发”完全是两码事，下面把这次实践记一下，代码都是能直接跑的那种。

场景：别一上来就 gather 全部

先放个错误示范，我一开始大概就是这么写的：

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

这段代码不是不能用，问题是 urls 一多，等于你同时打开了和 urls 数量一样多的连接。对方服务没挂是人家脾气好，给你返回 429 也很合理。我一开始还以为是 aiohttp 的问题，后来把并发降到 10，世界立刻安静了。

用 Semaphore 控制并发数

最直接的办法是 asyncio.Semaphore。这玩意儿不复杂，就是拿个信号量卡住同时执行的协程数量：

sem = asyncio.Semaphore(10)

async def fetch(session, url):
    async with sem:
        async with session.get(url) as resp:
            if resp.status == 429:
                retry_after = resp.headers.get("Retry-After", "1")
                await asyncio.sleep(float(retry_after))
                raise RuntimeError("rate limited")
            resp.raise_for_status()
            return await resp.json()

这里注意 async with sem 包住的是整个请求过程，不是只包住创建 task。你如果只包住 session.get 的创建，后面读响应体的时候信号量已经放掉了，并发还是会飘。除了信号量，aiohttp.TCPConnector(limit=10) 也建议加上，算是双保险：

connector = aiohttp.TCPConnector(limit=10)
async with aiohttp.ClientSession(connector=connector) as session:
    ...

Semaphore 管的是协程层面的并发，TCPConnector 管的是连接池，两个不是一回事，但都加上心里踏实。

再上一层：按每秒请求数限流

有些接口不限制你同时开几个连接，它限制的是每秒多少次。这时候只靠 Semaphore 还不够，因为 10 个协程一秒钟可能跑完三轮。我这次用的是 aiolimiter，装一下：

pip install aiolimiter

用法不复杂：

from aiolimiter import AsyncLimiter

limiter = AsyncLimiter(max_rate=5, time_period=1)

async def fetch(session, url):
    async with limiter:
        async with session.get(url, timeout=10) as resp:
            if resp.status == 429:
                await asyncio.sleep(1)
                return await fetch(session, url)
            resp.raise_for_status()
            return await resp.json()

max_rate=5, time_period=1 就是每秒最多 5 个请求。这个限制和 Semaphore 不冲突，我的做法是 Semaphore 限制“同时在路上”的请求数，AsyncLimiter 限制“每秒发出去”的请求数。两个都加上之后，接口那边的 429 基本就没了。如果不想多一个依赖，也能自己写个令牌桶，但说实话，工具库已经写得挺稳了，没必要为了省一个依赖手搓，除非你有特殊需求。

重试别写成死循环

429 和 5xx 都要重试，但重试次数得有个上限，退避也得有。我一般是 tenacity 或者自己写个简单循环：

async def fetch_with_retry(session, url, retries=3):
    for attempt in range(retries):
        try:
            async with limiter:
                async with session.get(url, timeout=10) as resp:
                    if resp.status == 429:
                        await asyncio.sleep(2 ** attempt)
                        continue
                    resp.raise_for_status()
                    return await resp.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)

这里 2 ** attempt 就是 1、2、4 秒，别一上来就固定 sleep(1)，也别在 429 的时候疯狂重试，不然限流层会认为你是在挑衅它。重试的时候最好只重试幂等请求，这个看接口文档，别把写操作重试出问题来。

最后落地的配置

我最后把参数收在一处，方便改：

CONCURRENCY = 10
RPS = 5
TIMEOUT = 10

sem = asyncio.Semaphore(CONCURRENCY)
limiter = AsyncLimiter(RPS, 1)
connector = aiohttp.TCPConnector(limit=CONCURRENCY)
timeout = aiohttp.ClientTimeout(total=TIMEOUT)

然后 main 里：

async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
    results = await asyncio.gather(*[fetch_with_retry(session, u) for u in urls])

跑起来之后，日志不再刷屏，429 也基本看不到了，速度虽然比“全量并发”慢一点，但至少能稳定跑完。异步编程不是把 await 一加就完事，限流也不是为了好看，主要是别把对方打挂，也别把自己拖进无穷无尽的重试里。不出问题的话，应该就能稳定跑完了。