用 aiohttp 实现带并发信号量限制的批量抓取器

抓 500 个 URL，串行 requests 要跑七八分钟；直接 asyncio.gather(*tasks) 全量并发又容易被目标站限流、被本地文件描述符耗尽打回 OSError: [Errno 24] Too many open files。真正可用的做法是：协程模型 + asyncio.Semaphore 限流 + aiohttp.TCPConnector 限制连接数 + 超时与重试。本文给出一份能直接跑起来的完整实现，并把限流环节里常见的坑（信号量写错位置、ClientSession 被提前关闭、超时叠加后实际耗时翻倍）逐个拆开讲。

为什么是 aiohttp + Semaphore，而不是线程池

先明确两组数字的差别。抓 N 个页面，平均单请求耗时 300ms（含 RTT 与 TLS 握手摊销），那么：

方案	并发度	总耗时估算	主要瓶颈
requests 串行	1	500 × 0.3s ≈ 150s	网络等待
ThreadPoolExecutor(max_workers=50)	50	≈ 3s + 线程开销	线程栈内存、GIL 外的调度
aiohttp + Semaphore(50)	50	≈ 3s	连接池与目标站限流阀值

线程池方案在 50 并发时每个线程默认 8MB 虚拟栈，500 并发就很难看了；而且线程创建/销毁本身就是同步开销。协程方案在单线程事件循环里切换，1 万个并发任务只占几 MB 堆内存。真正的约束不在客户端，而在目标站点和你本机的文件描述符上限——所以不管用什么模型，并发度都必须被显式限制住。

这里的分层是清楚的：asyncio.Semaphore 控制「同时在飞的请求数」；TCPConnector(limit=N) 控制「同时打开的 TCP 连接数」，包括 keep-alive 复用后的空闲连接；ClientTimeout 控制单个请求的生命周期。三者职责不同，缺一个都会出问题。

最小可用版本：50 行跑通限流抓取

先看骨架，后面再逐块加固。注意 session.get() 必须用 async with 包住，否则响应体没读完连接不会归还连接池。

import asyncio
import aiohttp

URLS = [f"https://httpbin.org/delay/1?id={i}" for i in range(20)]
CONCURRENCY = 5

async def fetch(session: aiohttp.ClientSession, sem: asyncio.Semaphore, url: str):
    async with sem:  # 关键：在整个请求生命周期内持锁
        async with session.get(url) as resp:
            body = await resp.text()
            return url, resp.status, len(body)

async def main():
    sem = asyncio.Semaphore(CONCURRENCY)
    connector = aiohttp.TCPConnector(limit=CONCURRENCY)
    timeout = aiohttp.ClientTimeout(total=10, connect=5)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [asyncio.create_task(fetch(session, sem, u)) for u in URLS]
        for coro in asyncio.as_completed(tasks):
            url, status, size = await coro
            print(f"{status} {size:>7}B {url}")

if __name__ == "__main__":
    asyncio.run(main())

跑起来会看到每 5 个一批地输出，总耗时略大于 4 秒（20 个请求 ÷ 5 并发 × 1 秒延迟）。如果把 async with sem 写在 session.get() 外面但把 await resp.text() 放到锁外，你会看到连接数超限、TCPConnector 开始排队，限流形同虚设——这是最常见的写法错误。

错误写法与真实报错：信号量位置放错的三种形态

形态一：只锁住了”发起请求”

# 错误示例：锁的粒度太细
async def fetch_bad(session, sem, url):
    async with sem:
        resp = await session.get(url)      # 只锁到拿到响应头
    async with resp:
        return await resp.text()           # 读 body 时已经释放锁

结果：20 个请求的 body 读取阶段全部并发，连接池被瞬时打满，TCPConnector 开始丢弃或排队，日志里可能出现 RuntimeWarning: coroutine 'ClientSession._request' was never awaited（如果你忘了 await 又吞掉异常）或直接卡住。判断依据很简单：在 fetch 里打印 len(sem._value)，如果长期在 0 和 5 之间剧烈震荡而不是稳定为「并发数 − 在飞数」，说明锁没包住完整生命周期。

形态二：Session 在任务外被关闭

# 错误示例
session = aiohttp.ClientSession()
tasks = [asyncio.create_task(fetch(session, sem, u)) for u in URLS]
await session.close()          # 任务还没跑完就关了
await asyncio.gather(*tasks)

典型报错：RuntimeError: Session is closed 或 aiohttp.client_exceptions.ClientConnectionError: Connection closed。正确做法是 async with aiohttp.ClientSession(...) as session: 把「创建任务」和「等待任务」都放在 with 块内部。

形态三：每个请求新建 Session

这是性能陷阱而非报错。每次 ClientSession() 都意味着独立的连接池、独立的 DNS 缓存，TLS 握手无法复用。抓 500 个同域 URL 时，这种写法比串行还慢。一个进程内一个 Session 对应一组相同的连接配置，跨域抓取时按域名分组建 Session 即可。

加固版实现：重试、状态码校验、结果落盘

生产可用的抓取器还需要处理：5xx 与 429 的退避重试、非 2xx 的跳过、单条异常不拖垮整体（return_exceptions=True 或任务内 try/except）、以及限速配合。下面这份是完整实现，可直接替换 URLS 使用。

import asyncio
import csv
import random
from dataclasses import dataclass, field
from typing import Iterable

import aiohttp

@dataclass
class Result:
    url: str
    status: int = 0
    size: int = 0
    error: str = ""
    attempts: int = 0

RETRY_STATUS = {429, 500, 502, 503, 504}

async def fetch_one(
    session: aiohttp.ClientSession,
    sem: asyncio.Semaphore,
    url: str,
    max_retries: int = 3,
) -> Result:
    res = Result(url=url)
    for attempt in range(1, max_retries + 1):
        res.attempts = attempt
        try:
            async with sem:
                async with session.get(url, allow_redirects=True) as resp:
                    body = await resp.read()
                    res.status = resp.status
                    res.size = len(body)
                    if resp.status in RETRY_STATUS and attempt < max_retries:
                        raise aiohttp.ClientResponseError(
                            resp.request_info, resp.history,
                            status=resp.status, message="retryable",
                        )
                    return res
        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
            res.error = f"{type(exc).__name__}: {exc}"
            if attempt == max_retries:
                return res
            # 指数退避 + 抖动，避免同批任务同时重试形成尖峰
            backoff = 0.5 * (2 ** (attempt - 1)) + random.uniform(0, 0.3)
            await asyncio.sleep(backoff)
    return res

async def crawl(urls: Iterable[str], concurrency: int = 50) -> list[Result]:
    sem = asyncio.Semaphore(concurrency)
    connector = aiohttp.TCPConnector(
        limit=concurrency,          # 总连接数上限
        limit_per_host=10,          # 单域上限，防打死一个站
        ttl_dns_cache=300,          # DNS 缓存 5 分钟
        enable_cleanup_closed=True, # 清理自 3.8 起遗留的 closed transport
    )
    timeout = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10)
    headers = {"User-Agent": "fwdd-crawler/1.0 (+https://fwdd.net)"}

    results: list[Result] = []
    async with aiohttp.ClientSession(
        connector=connector, timeout=timeout, headers=headers
    ) as session:
        tasks = [asyncio.create_task(fetch_one(session, sem, u)) for u in urls]
        for fut in asyncio.as_completed(tasks):
            results.append(await fut)
    return results

def dump(results: list[Result], path: str = "crawl.csv") -> None:
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["url", "status", "size", "attempts", "error"])
        for r in results:
            writer.writerow([r.url, r.status, r.size, r.attempts, r.error])

if __name__ == "__main__":
    targets = [f"https://httpbin.org/status/200?i={i}" for i in range(100)]
    out = asyncio.run(crawl(targets, concurrency=20))
    ok = sum(1 for r in out if 200 <= r.status < 300)
    print(f"total={len(out)} ok={ok}")
    dump(out)

几处参数值得单独说明。limit_per_host 设为总并发的一半以下比较稳妥，很多站点的 WAF 对单 IP 每秒请求数敏感，而 limit 只管总量不管分布。ttl_dns_cache=300 让同域批量抓取只做一次 DNS 解析；enable_cleanup_closed=True 是为了绕开部分服务端不返回 TLS close_notify 导致的连接泄漏报错 SSL error ... APPLICATION_DATA_AFTER_CLOSE_NOTIFY。

退避里的 random.uniform(0, 0.3) 抖动不能省。去掉之后，一批任务会在同一时刻集体重试，瞬间又打满目标站，重试反而成了放大器。

调参与观测：并发数到底开多少

没有万能值，但有明确的约束链。先用下面两行确认本机上限：

ulimit -n                 # macOS/Linux 默认常见 256 或 1024
python -c "import resource,sys; print(resource.getrlimit(resource.RLIMIT_NOFILE))"

每个在飞的连接占 1 个 fd，加上 DNS 的 UDP socket 与日志文件，实际可用并发约为 ulimit -n 的 80%。看到 OSError: [Errno 24] Too many open files 就是撞了这条线，先 ulimit -n 8192 再谈并发数。

其次是客户端侧的经验区间：

• 同域小站/个人站：2~5，超过就是压测行为。
• 普通站点批量拉页面：10~20，配 limit_per_host=5。
• API 或 CDN 类目标：50~100，前提是对方文档给了 QPS 额度。
• 本机 CPU 密集的后处理：并发数别超过 os.cpu_count() 的 2 倍，否则事件循环被解析逻辑阻塞，限流数字再大也没用。

观测手段不用上 APM，跑完统计 attempts > 1 的比例就够了：超过 10% 说明并发过高或对方在限流，往下调一档再跑；error 列里大量 asyncio.TimeoutError 则说明 total=15 太紧，需要放宽或减少并发。

与速率限制配合：Semaphore 之外再加一层令牌桶

信号量约束的是「同时在飞的数量」，不是「每秒发起数」。抓一个对 QPS 敏感的目标，你需要的是每秒最多 N 个请求，此时信号量和 asyncio.sleep 都要上，或者直接用一个最小间隔锁：

class RateLimiter:
    def __init__(self, rate_per_sec: float):
        self.min_interval = 1.0 / rate_per_sec
        self._lock = asyncio.Lock()
        self._next_at = 0.0

    async def acquire(self) -> None:
        async with self._lock:
            loop = asyncio.get_running_loop()
            now = loop.time()
            wait = self._next_at - now
            self._next_at = max(now, self._next_at) + self.min_interval
        if wait > 0:
            await asyncio.sleep(wait)

# 用法：在 fetch_one 内部，sem 之前
# limiter = RateLimiter(rate_per_sec=8)
# await limiter.acquire()
# async with sem: ...

_next_at 的自增写法保证了在锁内取号、锁外等待，多个协程排队时不会因为多次 asyncio.sleep 的调度误差而挤在一起。8 QPS 配 Semaphore(20)，就是「每秒最多放 8 个进去，最多 20 个同时在飞」。

顺序很重要：先过速率限制再过信号量。反过来会让信号量名额被 sleep 中的协程白占，实际吞吐低于配置值。

小结

一套能上生产的限流抓取器由四层构成：Semaphore 管并在飞数量、TCPConnector(limit, limit_per_host) 管连接与域名分布、ClientTimeout(total, connect, sock_read) 管单项耗时上限、RateLimiter 管发起速率。信号量必须包住从 session.get() 到 resp.read() 的完整生命周期，ClientSession 必须由 async with 管理且只有一组配置一个实例。调试时优先看 attempts 分布与 fd 上限，八个异常里有七个能靠这两项定位。

FAQ

Q1：报 RuntimeError: Event loop is closed 怎么办？

通常是把 asyncio.run() 套在了已经运行的事件循环里（例如 Jupyter、或在 async 函数中再调 asyncio.run）。Jupyter 环境下改为 await crawl(...) 直接调用；脚本环境确认只有一个 asyncio.run() 入口，且 ClientSession 在 with 块内关闭。

Q2：明明设了 Semaphore(50)，实测并发只有 10 左右？

检查 TCPConnector(limit=...) 是否也设了 50，以及 limit_per_host 是否过小。另一个常见原因是目标站 HTTP/1.1 连接复用下，浏览器端的连接数由 limit_per_host 决定，默认值 0 表示不限制，但若显式写成 5 就会成为实际瓶颈。

Q3：asyncio.TimeoutError 和 aiohttp.ServerTimeoutError 有什么区别？

aiohttp.ServerTimeoutError 是 aiohttp 自己对 ClientTimeout 各阶段（connect / sock_read / total）判定的超时，继承自 ClientError；asyncio.TimeoutError 多来自外层 asyncio.wait_for 包装。捕获时两个都要写，否则重试逻辑会漏掉一类失败。

Q4：抓 HTTPS 时出现 SSLCertVerificationError 但浏览器正常？

Python 的证书链依赖 certifi，系统装了企业根证书也不会自动生效。临时方案是 aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False))，仅限内网调试；正解是 pip install --upgrade certifi 或通过 ssl.create_default_context(cafile="ca.pem") 显式传入公司根证书。

Q5：内存占用随 URL 数线性上涨，几万个任务怎么处理？

asyncio.create_task 一次性建几万个任务本身可行，但每个 task 都有对象开销与 traceback 缓存。数量超过一万时改用分批：每 2000 个 URL 一批 asyncio.gather 后再进下一批，或者用 asyncio.Queue 配固定数量的 worker 协程消费，内存曲线会变成常数。

Q6：需要抓的站需要 Cookie 登录态，Session 怎么共享？

ClientSession(cookies={...}) 初始化即可，Session 会自动维护 Set-Cookie 回写。多账号并发时按账号建 Session，不要在请求级别传 cookies= 覆盖，否则不同账号的 Cookie 会互相污染连接池里的复用连接。