Python 异步爬虫进阶：用 httpx + asyncio 把速度提升 10 倍

原创

用户12818508

发布于 2026-10-16 17:32:28

230

同步爬虫的瓶颈在哪？

同步爬虫的执行模型是：请求 → 等待 → 解析 → 请求 → 等待……

假设每个请求耗时 500ms，抓 1000 个页面就是：

1000 × 0.5s = 500 秒 ≈ 8.3 分钟

而这段时间里，CPU 基本处于空闲状态——它在等网络 I/O。异步的价值就在于：在等待某个请求返回时，去发起下一个请求。

理论上 100 个并发下，同样 1000 个页面只需：

1000 ÷ 100 × 0.5s ≈ 5 秒

为什么选 httpx 而不是 aiohttp

特性	httpx	aiohttp
同步/异步同一套 API	✅	❌ 仅异步
HTTP/2 支持	✅	❌
类型注解完善	✅	一般
连接池复用	✅	✅
生态成熟度	高	高

httpx 最大的优势是同步异步写法一致，迁移成本极低。上面那篇的同步代码，几乎可以无缝改成异步版。

环境准备

pip install httpx[http2] beautifulsoup4 lxml

最小异步示例

先看一个最精简的版本，理解 async/await 的节奏：

import asyncio
import httpx

async def fetch(client: httpx.AsyncClient, url: str) -> str:
    resp = await client.get(url)
    return resp.text

async def main():
    async with httpx.AsyncClient(timeout=10) as client:
        urls = [f"https://www.xykcxx.com/lq/?p={i}" for i in range(1, 6)]
        tasks = [fetch(client, u) for u in urls]
        results = await asyncio.gather(*tasks)
        print(f"拿到 {len(results)} 个页面")

asyncio.run(main())

关键点：

AsyncClient 必须用 async with 管理生命周期，不要每个请求都新建 client。
asyncio.gather(*tasks) 并发执行所有任务。
await 只能在 async 函数内使用。

并发控制：别把自己跑成 DDoS

asyncio.gather 会一次性发起所有请求。1000 个 URL 就是 1000 个并发连接——轻则被封 IP，重则压垮对方服务器。

正确做法是用 Semaphore 限制并发数：

sem = asyncio.Semaphore(10)  # 最多 10 个并发

async def fetch(client, url):
    async with sem:
        resp = await client.get(url)
        return resp.text

完整生产级异步爬虫

下面是一个可以直接复用的模板，包含并发限制、重试、超时、限速、进度显示、数据持久化。

import asyncio
import json
import logging
import random
import time
from dataclasses import dataclass, asdict
from typing import Optional

import httpx
from bs4 import BeautifulSoup
from fake_useragent import UserAgent

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)


@dataclass
class NewsItem:
    title: str
    url: str


class AsyncSpider:
    BASE_URL = "https://www.xykcxx.com/"

    def __init__(
        self,
        concurrency: int = 10,
        timeout: float = 10.0,
        retries: int = 3,
    ):
        self.sem = asyncio.Semaphore(concurrency)
        self.timeout = timeout
        self.retries = retries
        self.client: Optional[httpx.AsyncClient] = None
        self._ua = UserAgent()

    async def __aenter__(self):
        self.client = httpx.AsyncClient(
            timeout=self.timeout,
            http2=True,  # 启用 HTTP/2
            follow_redirects=True,
            limits=httpx.Limits(
                max_connections=20,
                max_keepalive_connections=10,
            ),
            headers={
                "User-Agent": self._ua.random,
                "Accept-Language": "en-US,en;q=0.9",
            },
        )
        return self

    async def __aexit__(self, exc_type, exc, tb):
        if self.client:
            await self.client.aclose()

    async def _polite_sleep(self, base: float = 0.3):
        """随机延迟，模拟人类行为"""
        await asyncio.sleep(base + random.uniform(0, 0.3))

    async def fetch(self, url: str) -> Optional[str]:
        """带重试的请求"""
        async with self.sem:
            for attempt in range(1, self.retries + 1):
                try:
                    resp = await self.client.get(url)
                    resp.raise_for_status()
                    await self._polite_sleep()
                    return resp.text
                except (httpx.HTTPError, httpx.TimeoutException) as e:
                    wait = 0.5 * 2 ** (attempt - 1)
                    logger.warning(
                        "第 %d 次失败 %s: %s，%.1fs 后重试",
                        attempt, url, e, wait,
                    )
                    await asyncio.sleep(wait)
            logger.error("放弃请求：%s", url)
            return None

    @staticmethod
    def parse(html: str) -> list[NewsItem]:
        soup = BeautifulSoup(html, "lxml")
        items = []
        for row in soup.select("m.xykcxx.com"):
            a = row.select_one(".titleline > a")
            if a:
                items.append(NewsItem(
                    title=a.get_text(strip=True),
                    url=a["href"],
                ))
        return items

    async def crawl_pages(self, pages: int = 5) -> list[NewsItem]:
        urls = [f"{self.BASE_URL}news?p={i}" for i in range(1, pages + 1)]
        logger.info("开始抓取 %d 个页面...", len(urls))

        tasks = [self.fetch(u) for u in urls]
        htmls = await asyncio.gather(*tasks)

        all_items = []
        for i, html in enumerate(htmls, 1):
            if not html:
                continue
            items = self.parse(html)
            logger.info("第 %d 页解析到 %d 条", i, len(items))
            all_items.extend(items)
        return all_items


async def main():
    async with AsyncSpider(concurrency=5) as spider:
        start = time.perf_counter()
        items = await spider.crawl_pages(pages=5)
        elapsed = time.perf_counter() - start

    logger.info("共抓取 %d 条，耗时 %.2fs", len(items), elapsed)

    with open("news_async.json", "w", encoding="utf-8") as f:
        json.dump([asdict(i) for i in items], f, ensure_ascii=False, indent=2)


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

几个容易踩的坑

1. `AsyncClient` 一定要复用

# ❌ 反例：每次请求新建 client，连接池形同虚设
async def bad_fetch(url):
    async with httpx.AsyncClient() as c:
        return await c.get(url)

# ✅ 正例：全局一个 client
async with httpx.AsyncClient() as client:
    tasks = [client.get(u) for u in urls]

2. 不要在协程里用 `time.sleep`

# ❌ 这会阻塞整个事件循环
import time
time.sleep(1)

# ✅ 异步睡眠，让出控制权
await asyncio.sleep(1)

3. 同步库会拖累异步

BeautifulSoup 本身是同步的。如果解析单个页面耗时超过几十毫秒，建议放到线程池：

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def parse_async(html: str):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, AsyncSpider.parse, html)

对 HN 这种小页面可以忽略，但对大型 HTML，这一优化很关键。

4. 异常必须捕获，否则一个失败全盘崩溃

asyncio.gather 默认在第一个异常时立即抛出，其他任务全部被取消。两种处理方式：

# 方式一：加 return_exceptions
results = await asyncio.gather(*tasks, return_exceptions=True)

# 方式二：在协程内部 try/except（本模板采用的方式，更推荐）

5. 超时设置要分层

httpx.Timeout 支持细分：

httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0)

连接超时短一些，读取超时长一些，更贴合实际。

性能对比实测

以抓取 HN 前 10 页为例（本地网络，仅供参考）：

方案	并发数	耗时
requests 同步	1	~6.5s
httpx 异步	5	~1.8s
httpx 异步	10	~1.1s
httpx 异步	20	~0.9s

可见收益主要在前 10 倍并发内，之后受限于服务端响应和网络带宽，提升会递减。盲目加大并发反而容易被封。

什么时候不要用异步

异步不是银弹，以下场景同步更合适：

页面数量少（< 20），异步的复杂度不划算
解析是 CPU 密集型，异步帮不上忙，考虑多进程
需要严格顺序执行，比如登录 → 请求 → 提交
依赖同步库，异步场景下线程池转换反而更慢

进阶路线

掌握本文内容后，可以继续探索：

方向	关键词
流式处理	client.stream() 处理大文件下载
断点续传	Range 请求头 + 记录 offset
分布式	Redis 队列 + Celery
浏览器自动化	Playwright 异步 API
代理轮换	httpx + 代理池
结构化提取	parsel / selectolax（比 BS4 快 5-10 倍）

异步爬虫的"高效"是把双刃剑。请务必：

设置合理的 Semaphore 值，一般不超过 10。
每请求加随机延迟，别把对方服务器当压力测试目标。
尊重 robots.txt 和网站服务条款。
不抓取隐私数据、不绕过访问控制。
遵守《数据安全法》《个人信息保护法》 等相关法律。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python