Python异步爬虫实战：aiohttp与asyncio把采集吞吐提升十倍

引言：同步爬虫的天花板不在 CPU，而在等待

绝大多数人写的第一版爬虫长这样：一个 for 循环，里面 requests.get(url)，拿到 HTML 解析、入库，然后进入下一个 URL。逻辑清晰，调试友好，跑一百个页面毫无压力。但当采集目标从一百个变成十万个时，问题立刻暴露：单请求平均耗时 280ms，其中 DNS 解析 20ms、TCP+TLS 握手 80ms、等待服务端响应 170ms、传输 10ms——真正消耗 CPU 的时间不到 5ms，其余 275ms 里，你的进程只是在 recv() 上阻塞，什么也没干。

换句话说，同步爬虫的瓶颈从来不是算力，而是一次只能等一件事。CPU 在 95% 以上的时间里处于空闲，你付了 100% 的机器成本，却只用了不到 5% 的计算能力。这就是为什么单纯给爬虫加机器、加 CPU 核数，几乎看不到吞吐提升的原因——瓶颈压根不在那里。

有人会说：上线程池啊。ThreadPoolExecutor(max_workers=50) 确实能把吞吐拉上去，这是最省事的方案，本文也不否认它在很多场景下够用。但线程池有三个绕不开的成本：一是每个线程要预留栈空间（默认 8MB 虚拟内存），几百个线程就开始吃内存；二是线程切换由内核调度器主导，上下文切换在万级 QPS 下变成可观开销；三是并发度难以做细粒度的动态调节——你没法在请求级别上优雅地插一个"每域名最多 8 个并发"的闸门。

asyncio + aiohttp 提供的正是另一种抽象：单线程、事件驱动、在 await 处主动让出控制权。协程的内存开销在 KB 级别，调度完全在用户态完成，并且并发控制可以精确到单个 await 语句。本文从原理讲到生产级实现，给出一套可以直接落地的异步采集骨架。

一、核心原理：asyncio 到底省掉了什么

1.1 事件循环与协程调度

asyncio 的核心是一个 EventLoop，本质是一个 select/poll/epoll/kqueue 的封装。当你写 await session.get(url) 时，发生的是：

关键点在于：挂起和唤醒都是用户态的栈帧操作，没有系统调用，没有线程创建，没有锁。这就是为什么单线程能扛住上千个并发连接——不是因为单线程更快，而是因为它把"等待"这件事的成本降到了近乎为零。

如果把线程模型和协程模型放在一起对比，差别会更加直观。一个线程从创建到销毁，要经历内核分配 task_struct、分配栈、加入调度队列、被抢占或主动让出、上下文切换保存寄存器等一整套流程；而一个协程的"挂起"本质上只是把一个 Python 帧对象的引用放进事件循环的回调队列，唤醒则是从队列里取出来继续 send()。前者的开销在微秒到毫秒量级，后者的开销在亚微秒量级，差了两到三个数量级。

还要澄清一个常见误解：asyncio 不是多线程，也不是并行。它做到的是"并发"——在单核上交错执行多个任务，让 I/O 等待时间被其他任务填满。如果你的爬虫里混入了大量 CPU 密集的解析逻辑（比如正则回溯、大型 JSON 反序列化），这些代码会阻塞整个事件循环，导致所有其他协程一起卡住。这是异步爬虫最隐蔽的性能陷阱之一，后文 FAQ 会专门讨论。

1.2 aiohttp 的连接池才是吞吐的真正来源

很多人以为异步的收益来自"同时发请求"，其实更重要的收益来自连接复用。一次 HTTPS 请求里，TCP 三次握手 + TLS 1.3 握手通常占 60~100ms，如果每个请求都重新建连，并发再高也是浪费。

aiohttp.ClientSession 内部维护一个 TCPConnector，默认 limit=100（总连接数上限）、limit_per_host=0（单域名不限）、开启 keep-alive。这意味着对同一域名的第 2 个到第 N 个请求，可以直接复用已建立的 TLS 连接，把 280ms 压到 180ms 左右。这也是为什么 ClientSession 必须全局复用，绝不能每个请求 new 一个——那样连接池形同虚设，还会因为 fd 泄漏把进程拖死。

连接池的另一个好处是它天然承担了"背压"的一部分职责。当并发请求数超过 limit 时，后续的 session.get() 不会立即发出，而是挂起等待连接释放。这相当于在传输层给系统装了一个隐形的限流阀，避免瞬间打满 fd 上限或者被目标站点的 WAF 判定为攻击。

需要特别注意 limit_per_host 这个参数。默认值 0 表示"不限制单域名并发"，看起来很美，但对单站点采集时极其危险：你可能一瞬间对同一个域名发起几百个连接，轻则被限流返回 429，重则直接触发 IP 封禁。生产环境里，永远显式设置 limit_per_host，这是一个几乎零成本、收益极高的保险。

1.3 用 Little’s Law 估算你的吞吐上限

排队论里的利特尔法则给出一个粗糙但好用的模型：

吞吐 X (req/s) = 并发数 N / 平均延迟 L (s)

• 串行：N=1，L=0.28s → X ≈ 3.6 req/s，十万 URL 需要 7.7 小时；
• 异步 N=200，若 L 因排队上升到 0.62s → X ≈ 322 req/s，十万 URL 约 5.2 分钟。

注意后半段：延迟 L 不是常数。并发上去后，服务端排队、带宽打满、DNS 阻塞都会让 L 上升，所以吞吐不会随 N 线性增长，最终会撞到某个拐点。调优的目标不是把并发开到最大，而是找到 L 开始非线性上升之前的那个 N。

这个模型最大的价值在于：它把"要不要加并发"变成了一个可以量化的问题。你只需要做一轮简单的压测——分别用 N=50、100、200、400 跑固定数量的 URL，记录实际耗时和成功数——就能画出自己环境下的吞吐曲线。如果 N 从 200 加到 400，吞吐只提升了 10%，说明你已经过了拐点，继续加并发只是在制造失败请求。

二、最小可用骨架：从 requests 迁移到 aiohttp

先看一个能跑起来的最小版本，重点观察三处差异：会话复用、并发聚合、超时配置。

import asyncio
import time
import aiohttp

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

async def fetch(session: aiohttp.ClientSession, url: str):
# 注意：必须在 async with 内部读完 body，否则连接不会归还连接池
async with session.get(url) as resp:
resp.raise_for_status()
html = await resp.text()
return url, len(html)

async def main():
connector = aiohttp.TCPConnector(
limit=200, # 全局连接数上限
limit_per_host=20, # 单域名并发上限，保护目标站点也保护自己
ttl_dns_cache=300, # DNS 结果缓存 5 分钟，避免重复解析
enable_cleanup_closed=True,
)
timeout = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10)

t0 = time.perf_counter()
async with aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers={"User-Agent": "Mozilla/5.0 (compatible; MyCrawler/1.0)"},
) as session:
tasks = [asyncio.create_task(fetch(session, u)) for u in URLS]
results = await asyncio.gather(*tasks, return_exceptions=True)

ok = sum(1 for r in results if not isinstance(r, Exception))
print(f"{ok}/{len(URLS)} 成功, 耗时 {time.perf_counter() – t0:.2f}s")

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

三个细节值得单独强调：

• return_exceptions=True：不加这个参数，任何一个任务抛异常都会让 gather 整体失败，前面跑完的结果全部丢失。加了之后异常作为返回值出现，由调用方统一归类处理。
• async with 内读完 body：session.get() 返回时响应头已到，但 body 还在流里。提前退出 async with 会触发连接 release 而非 reuse，连接池里的连接会逐渐被"半消费"状态污染。
• 三层超时：connect 管握手，sock_read 管两次读之间的间隔，total 兜底。只设 total 会让慢速攻击型的服务端拖住连接直到总超时。

这个骨架有个致命问题：1000 个 URL 一次性 create_task，如果换成十万个，光 Task 对象就吃掉几百 MB，而且没有任何背压——生产者速度远大于消费者。生产环境必须换成队列模型。

顺便说一句迁移时的常见习惯：很多从 requests 过来的同学会保留 requests.get(url, timeout=10) 的写法，然后在 aiohttp 里只写 session.get(url) 而忘了超时。aiohttp 默认 total=300s，一个卡死的连接能挂你五分钟。超时不是可选项，是异步爬虫的生存前提——在同步模型里，一个慢请求只是拖慢一个请求；在异步模型里，一个不超时的慢请求会长期占据一个连接名额，最终把整个连接池耗光。

三、实战案例：带限流、重试、背压的生产级采集器

下面这套结构是笔者在多个采集项目中反复验证过的：固定数量的 worker 协程 + 有界队列 + 信号量限流 + 指数退避重试。

import asyncio
import random
import aiohttp
from collections import defaultdict
from dataclasses import dataclass

@dataclass
class Stats:
ok: int = 0
fail: int = 0
retried: int = 0

class AsyncCrawler:
def __init__(self, concurrency: int = 128, per_host: int = 8):
self.sem = asyncio.Semaphore(concurrency)
self.per_host = per_host
self.host_sems: dict[str, asyncio.Semaphore] = defaultdict(
lambda: asyncio.Semaphore(self.per_host)
)
self.stats = Stats()

async def fetch(self, session, url: str, max_retry: int = 3) –> str | None:
host = url.split("/")[2]
for attempt in range(max_retry):
# 信号量要包住"建连 + 读 body"的全过程
async with self.sem, self.host_sems[host]:
try:
async with session.get(url) as resp:
if resp.status == 429:
wait = float(resp.headers.get("Retry-After", 2 ** attempt))
await asyncio.sleep(wait)
continue
if resp.status >= 500:
raise aiohttp.ClientResponseError(
resp.request_info, resp.history, status=resp.status
)
resp.raise_for_status()
html = await resp.text()
self.stats.ok += 1
return html
except (aiohttp.ClientError, asyncio.TimeoutError):
self.stats.retried += 1
if attempt == max_retry – 1:
self.stats.fail += 1
return None
# 指数退避 + 抖动：避免所有失败请求在同一时刻集体重试
await asyncio.sleep(2 ** attempt + random.random())
return None

async def worker(self, session, queue: asyncio.Queue, results: list):
while True:
try:
url = queue.get_nowait()
except asyncio.QueueEmpty:
return
try:
html = await self.fetch(session, url)
if html is not None:
results.append((url, len(html)))
finally:
queue.task_done()

async def run(self, urls: list[str], workers: int = 64):
queue: asyncio.Queue = asyncio.Queue(maxsize=10000) # 有界队列形成背压
results: list = []

connector = aiohttp.TCPConnector(
limit=200, limit_per_host=self.per_host, ttl_dns_cache=300
)
timeout = aiohttp.ClientTimeout(total=15, connect=5, sock_read=10)

async with aiohttp.ClientSession(
connector=connector, timeout=timeout
) as session:
# 生产者：把 URL 塞进队列，队列满时自动 await 等待
async def producer():
for u in urls:
await queue.put(u)

prod_task = asyncio.create_task(producer())
workers_tasks = [
asyncio.create_task(self.worker(session, queue, results))
for _ in range(workers)
]
await prod_task
await asyncio.gather(*workers_tasks)

print(
f"ok={self.stats.ok} fail={self.stats.fail} "
f"retried={self.stats.retried} collected={len(results)}"
)
return results

这段代码里有几个关键设计点，值得逐条拆解：

第一，双信号量结构。self.sem 是全局并发闸门，self.host_sems[host] 是单域名闸门。两者用同一个 async with 一起进入，意味着一个请求必须同时拿到"全局名额"和"该域名名额"才能发出。这样可以避免对单一站点过度施压，同时又不会因为某个慢域名把全局并发名额全部占死。

第二，信号量包住了读 body 的过程。注意 async with self.sem 的位置是包在 async with session.get(url) 的外层，而不是只包住建连。因为连接在 body 读完之前是不会归还的，如果信号量只包住 session.get()，那么大量请求会在"持有连接、等待 body"的状态下突破并发限制。

第三，Retry-After 头优先于指数退避。遇到 429 时，如果服务端明确告诉你"X 秒后再来"，就老老实实等 X 秒；没有这个头时才用 2 ** attempt 的指数退避。这是对目标站点的基本尊重，也是避免被封的实用手段。

第四，退避加入随机抖动。2 ** attempt + random.random() 里的 random.random() 看似无关紧要，实则是防止"重试风暴"的关键。如果一批请求同时失败，没有抖动的话它们会在同一时刻集体重试，形成新的尖峰；加了抖动后，重试时间被打散，压力被摊平。

第五，有界队列即背压。asyncio.Queue(maxsize=10000) 让生产者在队列满时自动 await 阻塞，从而限制了内存占用。如果队列无界，生产者可以瞬间把百万 URL 塞进去，内存直接爆炸。

第六，固定 worker 数量而非按 URL 数量建 Task。worker 协程数量是常量（如 64 或 128），每个 worker 循环从队列取任务。这样即使 URL 有百万条，Task 对象数量也稳定在几十到几百，内存占用可控。

第七，queue.task_done() 放在 finally 里。即使 fetch 内部抛出未捕获异常，也保证队列计数正确，不会让 join() 永远挂住。虽然本例里用的是 get_nowait 轮询而非 join()，但保留这个习惯能让你在未来改成阻塞式消费时少踩一个坑。

四、常见问题（FAQ）

Q1：为什么我的异步爬虫比同步还慢？

最常见的原因是在协程里写了阻塞代码。典型场景包括：用 time.sleep() 而不是 await asyncio.sleep()；用 requests 而不是 aiohttp；在解析 HTML 时用了大量 CPU 密集操作（复杂的 XPath、正则回溯）；同步写文件、同步访问数据库。任何一次阻塞都会让整个事件循环停摆，所有并发瞬间退化成串行。排查方法很简单：用 asyncio 的 debug 模式（asyncio.run(main(), debug=True)），它会打印出执行时间超过 100ms 的回调，直接定位阻塞点。

Q2：gather 和 as_completed 该用哪个？

gather 适合"我关心全部结果，且结果顺序重要"的场景，它按传入顺序返回。as_completed 适合"谁先完成先处理谁"的流式场景，比如边采集边入库、边采集边写文件。当 URL 数量达到十万级时，gather 需要把所有结果攒在内存里才能返回，内存压力大；as_completed 可以边完成边消费，内存曲线更平缓。实战中推荐：中小规模用 gather，大规模用队列 + as_completed 或 worker 模式。

Q3：协程里能不能用多线程/多进程？

可以，但要分清边界。CPU 密集的解析任务可以丢进 loop.run_in_executor(ThreadPoolExecutor(), fn)，把计算挪出事件循环；如果解析特别重，可以用 ProcessPoolExecutor。但要注意：executor 里的函数不能是协程，也不能跨进程传不可序列化的对象。另外，run_in_executor 返回的是 future，需要 await，且它本身会占用一个线程，别在协程里滥用。

Q4：为什么明明设置了 limit=200，实际并发上不去？

检查三个地方：一是 limit_per_host 是否把单域名卡死了（比如 limit_per_host=8，那你对单站点的有效并发就是 8）；二是目标站点的响应是否变慢，导致连接被长时间占用；三是 sock_read 超时是否过短，导致大量请求在读到一半时被中断，连接反复重建。用 connector.limit 和实际 in-flight 数量对比，能快速定位。

Q5：aiohttp 报 “Connector is closed” 是什么原因？

几乎都是 ClientSession 的生命周期管理问题。典型写法是在函数里 async with aiohttp.ClientSession() as session，函数返回后 session 关闭，但还有协程持有它去发请求。正确做法是在 main 级别创建 session，通过参数传给所有 worker，等所有任务结束后再退出 async with。这就是本文骨架里 session 在最外层创建的原因。

Q6：异步爬虫还需要遵守 robots.txt 和限速吗？

需要，而且比同步更重要。异步让你很容易在几秒内向一个站点发出上千请求，这种"瞬时压力"对小型站点是灾难性的，也最容易触发封禁。建议：默认对单域名限速（比如每秒 5~10 个请求），尊重 Retry-After，遇到 403/429 立即降速。技术上可以用 asyncio.Semaphore 之外再加一个基于时间的令牌桶，或者直接依赖 limit_per_host 配合 sock_read 超时形成自然节流。

五、踩坑清单与优化建议

踩坑一：连接泄漏导致进程卡死。 如果 async with session.get(url) 内部因为异常提前退出，连接其实是可以正常归还的；但如果你用 resp = await session.get(url) 这种非上下文管理器的写法，一旦忘记 resp.release() 或 resp.close()，连接就永远挂在池子里。坚持用 async with 是最省心的做法。

踩坑二：DNS 解析成为隐藏瓶颈。 aiohttp 底层用 getaddrinfo，在部分系统上是同步阻塞调用，会短暂卡住事件循环。开启 ttl_dns_cache 能大幅缓解；如果域名数量巨大且分散，考虑用 aiodns 加速器（aiohttp 支持 AsyncResolver）。

踩坑三：asyncio.run() 只能调用一次。 在 Jupyter、某些 Web 框架、以及已有事件循环的环境里，直接调 asyncio.run() 会报 “event loop is already running”。此时应该用 await main() 或 loop.run_until_complete()，并注意不要嵌套创建事件循环。

踩坑四：日志和监控在异步下会失真。 多个协程交错打印日志，时间戳容易混淆；计数器的自增在单线程里虽然安全，但如果引入 run_in_executor 就会产生竞态。建议用结构化日志（带 url、attempt、elapsed 字段），并对 Stats 的更新集中在事件循环线程内完成。

优化建议一：把解析也异步化。 HTML 解析本身是 CPU 操作，无法"异步"，但你可以控制它在哪执行。简单解析直接在协程里做（只要足够快），重型解析丢进 executor。关键是别让一个页面的解析时间超过一个请求的耗时。

优化建议二：用 aiofiles 或批量写入处理落库。 同步的 open().write() 会阻塞事件循环。小规模用 aiofiles，大规模用"内存攒批 + 定时 flush"，让 I/O 不在关键路径上。

优化建议三：给整个采集任务加一个"看门狗"。 用 asyncio.wait_for(main(), timeout=3600) 包住主流程，防止某个 bug 让任务无限挂起。同时记录每个 worker 的 in-flight 数量和平均延迟，这些指标比总吞吐更能反映系统健康度。

结语

把同步爬虫改成异步，核心从来不是背几个 async/await 语法，而是理解"等待"这件事的真实成本，然后用事件循环把它摊平。

更多硬核网安与AI工具包，请扫码获取完整源码！
本文从 epoll 到连接池、从 Little’s Law 到队列背压、从最小骨架到生产级限流重试，给出了一套可以直接抄作业的结构。落地时请记住三句话：session 要复用、超时必配置、并发要限流。做到这三点，十倍吞吐只是起点，真正的收益是你终于可以按"站点"而不是按"机器"来规划采集能力。