Python asyncio 爬虫实战：并发、限速、重试和落库

一、为什么要用 asyncio 写爬虫

传统同步爬虫的瓶颈通常不在 CPU，而在网络等待。一个请求发出去后，大部分时间都在等服务器响应，如果用同步方式串行执行，CPU 会长时间空闲。

asyncio 的核心价值是：在等待网络 I/O 时切换到其它任务，让一个进程同时推进多个请求。

适合使用异步爬虫的场景：

页面数量较多。
单页解析逻辑不重。
请求耗时主要来自网络。
需要可控并发、限速和重试。

不适合的场景：

页面需要大量浏览器渲染。
解析过程是重 CPU 计算。
目标站点反爬规则复杂且不稳定。

二、基础结构

一个稳定的爬虫不要只写一个 fetch 函数，建议拆成几个明确阶段：

1	url source -> scheduler -> fetcher -> parser -> storage

每一层只关心自己的职责：

url source：生成待抓取 URL。
scheduler：控制并发、去重、重试。
fetcher：发送 HTTP 请求。
parser：解析页面并提取数据。
storage：写入文件、数据库或消息队列。

三、最小异步请求

常用组合是 asyncio + aiohttp。

import asyncio
import aiohttp


async def fetch(session, url):
    async with session.get(url, timeout=10) as resp:
        resp.raise_for_status()
        return await resp.text()


async def main():
    urls = [
        "https://example.com/page/1",
        "https://example.com/page/2",
        "https://example.com/page/3",
    ]

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


asyncio.run(main())

这段代码能跑，但还不够适合生产使用。主要问题是没有并发上限、没有重试、没有分批、没有失败记录。

四、控制并发

并发不是越大越好。过高并发会导致：

本机连接数耗尽。
DNS 或代理压力过大。
目标站点返回 429。
数据写入端被打爆。

可以用 asyncio.Semaphore 控制同一时刻的请求数量：

import asyncio
import aiohttp


sem = asyncio.Semaphore(10)


async def fetch_with_limit(session, url):
    async with sem:
        async with session.get(url, timeout=10) as resp:
            resp.raise_for_status()
            return await resp.text()

如果目标站点有明显限速要求，还需要在请求前加入节流：

async with sem:
    await asyncio.sleep(0.2)
    async with session.get(url, timeout=10) as resp:
        ...

五、重试和退避

网络请求失败很常见。需要区分：

可重试：超时、连接重置、临时 5xx。
不重试：404、参数错误、解析规则不匹配。

简单重试函数：

import asyncio
import random


async def retry(coro_factory, max_times=3):
    for attempt in range(1, max_times + 1):
        try:
            return await coro_factory()
        except Exception:
            if attempt == max_times:
                raise
            delay = 0.5 * attempt + random.random() * 0.2
            await asyncio.sleep(delay)

调用方式：

1	html = await retry(lambda: fetch_with_limit(session, url))

生产项目里建议记录每次失败原因，方便后续判断是网络问题、反爬问题，还是解析规则已经过期。

六、队列模型

当 URL 很多时，不建议一次性创建成千上万个任务。更稳的方式是使用 asyncio.Queue：

async def worker(name, queue, session):
    while True:
        url = await queue.get()
        try:
            html = await retry(lambda: fetch_with_limit(session, url))
            item = parse(html)
            await save(item)
        finally:
            queue.task_done()


async def run(urls):
    queue = asyncio.Queue()
    for url in urls:
        queue.put_nowait(url)

    async with aiohttp.ClientSession() as session:
        workers = [
            asyncio.create_task(worker(f"worker-{i}", queue, session))
            for i in range(10)
        ]
        await queue.join()

        for task in workers:
            task.cancel()

队列模型的好处是任务数量稳定，内存不会因为 URL 数量暴涨。

七、落库建议

爬虫落库要考虑重复写入和中断恢复。

建议字段：

url
url_hash
status
retry_count
content_hash
created_at
updated_at
error_message

关键点：

使用唯一索引避免重复写入。
每次请求记录状态。
解析结果和原始页面分开存。
失败 URL 可以单独导出重跑。

八、常见坑

常见问题：

忘记关闭 ClientSession。
把所有 URL 一次性 gather。
并发过高导致目标站点拒绝服务。
没有超时，任务一直挂住。
重试没有退避，失败时反而加大压力。
解析异常没有捕获，导致 worker 提前退出。

更稳的策略是先小规模跑通，再逐步增加并发，并持续观察成功率、耗时和失败分布。

Python asyncio 爬虫实战：并发、限速、重试和落库

Zihuatanejo