并发抓取：线程池与 asyncio

抓取慢通常不是 Python 慢，而是一次只等一个响应。本章先测单请求耗时、再由目标 QPS 反推并发数，然后给出线程池与 asyncio 两套可直接运行的实现。前置红线：遵守 robots.txt 与网站服务条款，控制请求频率与并发上限，不采集个人隐私与受版权保护的付费内容，不绕过登录、付费墙、验证码等访问控制，数据使用遵守《个人信息保护法》《数据安全法》《著作权法》。

先测量，再并发

把同一个 URL 串行请求 10 次，用 time.perf_counter() 记录每次耗时，看中位数与平均值：中位数代表「正常一次要多久」，平均值会被偶发慢请求拉高。拿到数字再反推并发：并发数 ≈ 目标 QPS × 单请求耗时。

单请求耗时	目标 QPS	理论并发	建议并发	说明
0.2s	5	1	2	串行就够，不必上并发
0.5s	10	5	4~6	线程池最舒服的区间
0.5s	20	10	6~8	先确认对方能承受

理论值只是起点。并发数只控制「同时在飞的请求」，不控制「每秒发出去多少请求」，所以两类限制要一起上：Semaphore（或 max_workers）控并发，令牌桶控速率。

线程池：`ThreadPoolExecutor`

用法三步：提交任务拿 Future、用 as_completed 按完成顺序取结果、按 URL 归集成败两份清单。max_workers 常见 4~8，IO 密集型可以略高于 CPU 核数，但真正决定上限的是目标站点的承受力，不是本机。

import json, threading, time
from concurrent.futures import ThreadPoolExecutor, as_completed
import httpx

URLS = [f"https://example.com/list?page={i}" for i in range(1, 201)]
HEADERS = {"User-Agent": "MyCrawler/1.0 (+https://example.com/bot)"}
_lock, _next_at = threading.Lock(), 0.0

def pace(min_interval: float = 0.5) -> None:
    """全局最小请求间隔（秒），多个线程共用一个时间片。"""
    global _next_at
    with _lock:
        now = time.monotonic()
        delay = max(0.0, _next_at - now)
        _next_at = max(_next_at, now) + min_interval
    if delay > 0:
        time.sleep(delay)

def fetch(client: httpx.Client, url: str) -> dict:
    pace()                                              # 先限速，再发请求
    resp = client.get(url)
    if resp.status_code == 429:                         # 被限流说明速率过高
        raise RuntimeError("429 too many requests")
    resp.raise_for_status()
    return {"url": url, "status": resp.status_code, "len": len(resp.text)}

def main() -> None:
    ok, failed = {}, {}
    with httpx.Client(timeout=10.0, headers=HEADERS, follow_redirects=True) as client:
        with ThreadPoolExecutor(max_workers=6) as pool:  # 并发上限 6，与速率无关
            futures = {pool.submit(fetch, client, u): u for u in URLS}
            for fut in as_completed(futures):
                url = futures[fut]
                try:
                    ok[url] = fut.result()
                except Exception as exc:                 # 单个失败不中断整批
                    failed[url] = f"{type(exc).__name__}: {exc}"
    json.dump(ok, open("ok.json", "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    json.dump(failed, open("failed.json", "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"完成：成功 {len(ok)}，失败 {len(failed)}")

if __name__ == "__main__":
    main()

asyncio：`aiohttp`

协程把等网络的时间让给别的任务，单线程就能跑出几十个并发，代价是整条链路必须异步，一处同步阻塞就会拖住事件循环。三层限制各管一段：asyncio.Semaphore 控任务并发、固定间隔控速率、TCPConnector(limit=20, limit_per_host=5) 控连接池。httpx 的等价写法是 httpx.AsyncClient(limits=httpx.Limits(max_connections=20, max_keepalive_connections=5))，语义一致，选型看团队熟悉度。

import asyncio, json, time
import aiohttp

URLS = [f"https://example.com/list?page={i}" for i in range(1, 201)]
HEADERS = {"User-Agent": "MyCrawler/1.0 (+https://example.com/bot)"}

_pace_lock, _next_at = asyncio.Lock(), 0.0

async def pace(min_interval: float = 0.5) -> None:
    """异步最小间隔：所有任务共用一个时间片，控的是速率而不是并发。"""
    global _next_at
    async with _pace_lock:
        now = time.monotonic()
        delay = max(0.0, _next_at - now)
        _next_at = max(_next_at, now) + min_interval
    if delay > 0:
        await asyncio.sleep(delay)                      # 锁外等待，不阻塞其他任务

async def worker(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore) -> dict:
    async with sem:                                     # 信号量控同时在飞的任务数
        await pace()                                    # 最小间隔控每秒请求数
        async with session.get(url) as resp:
            if resp.status == 429:
                raise RuntimeError("429 too many requests")
            resp.raise_for_status()
            return {"url": url, "status": resp.status, "len": len(await resp.text())}

async def main() -> None:
    sem = asyncio.Semaphore(5)                       # 并发 5；速率由 pace() 的 0.5 秒间隔决定
    connector = aiohttp.TCPConnector(limit=20, limit_per_host=5)     # 连接池层限流
    timeout = aiohttp.ClientTimeout(total=15)
    ok, failed = {}, {}
    async with aiohttp.ClientSession(connector=connector, timeout=timeout, headers=HEADERS) as s:
        tasks = [asyncio.create_task(worker(s, u, sem)) for u in URLS]
        results = await asyncio.gather(*tasks, return_exceptions=True)   # 个别失败不取消整批
    for url, res in zip(URLS, results):
        if isinstance(res, BaseException):
            failed[url] = f"{type(res).__name__}: {res}"
        else:
            ok[url] = res
    json.dump({"ok": ok, "failed": failed}, open("result.json", "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"完成：成功 {len(ok)}，失败 {len(failed)}")

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

两套写法怎么选

维度	`ThreadPoolExecutor`	`asyncio` + `aiohttp`	`httpx.AsyncClient`
并发模型	线程，等待网络时释放 GIL	单线程事件循环	单线程事件循环
常用并发上限	几十	几百到上千	几百到上千
控并发手段	`max_workers`	`asyncio.Semaphore`	`asyncio.Semaphore`
连接池限制	共用 `httpx.Client` 的连接池	`TCPConnector(limit=20, limit_per_host=5)`	`httpx.Limits(max_connections=20)`
适合场景	任务量中等、已有同步代码	高并发、全异步链路	想一套 API 兼顾同步与异步

GIL 与 IO 密集型

GIL 让同一时刻只有一个线程执行字节码，线程池跑不满多核。但网络请求绝大部分时间在等 socket，等待期间线程会释放 GIL，其他线程就能推进，这就是线程与协程对 IO 密集型有效的根本原因。解析 HTML、跑正则、算哈希不会释放 GIL，加线程只会互相抢锁，应该换 ProcessPoolExecutor，常见组合是「线程池抓取 + 进程池解析」。

失败归集、重跑与 429

并发必然伴随失败，关键是失败可追溯：按 URL 归集成败两份清单，成功项作为解析输入，失败项单独落盘，重跑只读失败清单——整批重跑既慢，又给对方凭空增加一倍流量。对自己可控的服务（预发环境、内网接口）可以压测，用 wrk -t4 -c100 -d60s 找容量拐点；对第三方站点没有压测资格，只能降频。并发数不是越大越好：超过对方承载会先收到 429 与 403，接着可能是连接被重置、IP 临时封禁，看到 429 的正确反应是降速并检查 Retry-After，而不是换代理继续打。

常见坑

每个任务新建客户端：连接数暴涨、握手开销大，应共用 Client 或 ClientSession。
在协程里调同步库：并发上不去、延迟反而变高，改用异步库或用 asyncio.to_thread 隔离。
只限并发不限速：对方响应变快时瞬时 QPS 飙升，信号量与令牌桶必须同时用。

小结：先用 10 次请求测出单请求耗时，按「并发数 ≈ 目标 QPS × 单请求耗时」反推并发，同时用信号量与令牌桶封住速率；线程池适合中等规模与同步代码，asyncio 适合高并发全异步链路，CPU 密集换多进程；按 URL 归集成败清单，收到 429 就降速而不是加代理。

笔记加载中…

标签：#Python3 #爬虫

返回知识库