# 使用 httpx、aiohttp 和代理进行异步 Python 抓取

异步可以将慢速抓取器变成快速抓取器——而快速抓取器会被阻止，除非你正确设置代理、信号量和超时。以下是使用 httpx 和 aiohttp 的完整模式，包括代码。

如果你的抓取器大部分时间都在等待网络响应，异步是提升速度的最大法宝——而代理是防止这种速度导致你被封禁的关键。本指南涵盖使用 `httpx` 和 `aiohttp` 的异步 Python 抓取以及代理的全流程：每个库如何设置代理，如何用信号量限制并发，如何设置真正生效的超时，以及如何在不破坏会话的情况下轮换 IP。所有这些都有可运行的代码和你可以替换的占位符凭据。

## 为什么选择异步，以及代理的作用

标准的 `requests` 是阻塞的：每次调用都要等响应后才能开始下一个。抓取 500 个页面，你就要连续支付 500 次往返。异步在一个事件循环中并发抓取它们，因此总时间趋向于最慢的单个请求而不是总和。问题是：从一个 IP 发出大量并发请求正是反爬虫系统监控的特征。解决方法不是放慢速度，而是通过一个[轮换代理池](https://quantumproxies.io/rotating-proxies)分散流量，并用信号量进行有意的节流。

## 在 httpx 中设置代理（异步）

`httpx` 是务实的默认选择，因为一个客户端模型同时支持同步和异步，并且支持 HTTP/2。注意现代 API：它是客户端上的一个单一 `proxy=` 参数，而不是旧的 `proxies=` 字典——这是升级后常见的“为什么我的代理被忽略”困惑的来源。凭据直接放入代理 URL 中。

```python
import asyncio, httpx

PROXY = "http://USER:PASS@residentialpremium.quantumproxies.io:8080"

async def fetch(client, url):
    r = await client.get(url, timeout=httpx.Timeout(20.0))
    return url, r.status_code, r.text

async def main(urls):
    async with httpx.AsyncClient(proxy=PROXY, http2=True) as client:
        tasks = [fetch(client, u) for u in urls]
        return await asyncio.gather(*tasks)

urls = ["https://httpbin.org/ip"] * 5
print(asyncio.run(main(urls)))
```

一个客户端，跨每个请求重复使用，这是关键——它保持连接池的温暖，因此你可以跳过通过代理的重复 TLS 握手。为每个请求创建一个新的客户端是最常见的异步性能错误：它在每次调用时抛弃了连接重用和 cookie 状态。

## 在 aiohttp 中设置代理（每个请求）

`aiohttp` 是 asyncio 原生的，提供最精细的并发控制，但其代理约定不同：代理是在 `session.get()` 上按请求传递的，而不是在会话上。这实际上对轮换很方便。这里有两点需要注意——默认的 User-Agent 字面上是 `Python/3.x aiohttp/3.x`，这是一个明显的信号你必须覆盖，并且你应该明确设置连接池大小。

```python
import aiohttp, asyncio

PROXY = "http://USER:PASS@residentialpremium.quantumproxies.io:8080"
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
           "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"}

async def fetch(session, url):
    timeout = aiohttp.ClientTimeout(total=30, connect=10, sock_read=20)
    async with session.get(url, proxy=PROXY, timeout=timeout) as resp:
        return url, resp.status, await resp.text()

async def main(urls):
    conn = aiohttp.TCPConnector(limit=20, limit_per_host=8)
    async with aiohttp.ClientSession(connector=conn, headers=HEADERS) as session:
        return await asyncio.gather(*(fetch(session, u) for u in urls))

print(asyncio.run(main(["https://httpbin.org/ip"] * 5)))
```

`TCPConnector(limit=20, limit_per_host=8)` 限制总的打开连接数和对任何单个主机的连接数——这是礼貌的第一道防线，防止一个目标吸收你的整个池。

![asyncio.gather 将所有抓取任务排队，asyncio.Semaphore(15) 让同时运行的任务不超过 15 个，轮换代理网关在并行请求到达目标之前为每个请求分配新的出口 IP。卡片总结：信号量负责限流，网关负责伪装，两者缺一，异步的速度就会变成一波封锁。](https://res.cloudinary.com/dmqliab0t/image/upload/f_auto,q_auto,w_1200/v1785335150/blog/generated/qp-b9-010-diag1.png)
*信号量节流；轮换网关伪装。你需要两者——没有它们的异步速度就是一个阻塞波。*

## 用信号量限制并发

无限制地发射并发请求是烧毁代理池和触发速率限制的最快方式。`asyncio.Semaphore` 是节流器：它限制一次在飞行中的请求数量，无论你排队了多少任务。设置一个全局限制，对于激进的任务也设置一个每个域的限制。

```python
sem = asyncio.Semaphore(15)  # never more than 15 requests in flight

async def guarded_fetch(client, url):
    async with sem:
        return await fetch(client, url)

async def main(urls):
    async with httpx.AsyncClient(proxy=PROXY) as client:
        return await asyncio.gather(*(guarded_fetch(client, u) for u in urls))
```

正确的数量取决于目标的容忍度和你的池大小，而不是你的机器能跑多快。保守地开始（10–20），观察你的阻塞率，并且只有在成功率保持高时才提高。在一个粘性 IP 上，保持在个位数。

## 超时：为每个阶段建模

代理请求比直接请求失败的地方更多——DNS 解析、连接到代理、隧道设置、连接到目标、等待头部和读取主体都是不同的停顿。单一的整体超时掩盖了哪个阶段挂起。`aiohttp` 的 `ClientTimeout(total=, connect=, sock_read=)` 和 `httpx` 的 `Timeout()` 允许你分别限制它们。唯一没有例外的规则：永远不要在没有超时的情况下发出请求，否则一个死锁将永远挂起一个协程并悄悄饿死你的事件循环。

## 在不破坏会话的情况下轮换代理

有两种轮换策略，选择错误会破坏你的数据。随机每请求轮换非常适合无状态页面抓取。但它会破坏任何依赖于 cookie、登录或本地化的流程，因为请求二落在与请求一不同的 IP 上。清晰的分割：在逻辑单元边界轮换——每个抓取段或每个账户一个 IP——并使用一个轮换网关，它会自动给你一个新的出口，这样你的代码永远不需要管理一个列表。

```python
# rotating gateway: one endpoint, new exit IP per request
ROT = "http://USER:PASS@residentialpremium.quantumproxies.io:8080"

# sticky session: same IP for a multi-step flow, tag the session id
STICKY = "http://USER-sid-a1b2-ttl-30m:PASS@residentialpremium.quantumproxies.io:8080"

async def crawl_segment(urls):
    async with httpx.AsyncClient(proxy=ROT) as client:  # rotates per call
        return await asyncio.gather(*(fetch(client, u) for u in urls))
```

轮换住宅网关是高并发抓取的务实默认选择：每请求轮换跨越 90M+ 个 IP 在 200+ 个国家中，当购物车或登录需要相同的出口几分钟时提供粘性会话。如果你在权衡住宅与 ISP 或数据中心的工作，我们的[代理类型选择指南](https://quantumproxies.io/blog/mobile-vs-residential-vs-isp-proxies-which-one-should-you-use-in-2026)列出了权衡。

[获取一个轮换住宅网关](https://quantumproxies.io/rotating-proxies)

## 重试、退避和抖动

死锁和瞬时阻塞在规模上是正常的，而不是例外。但天真的重试会使情况更糟：当 50 个异步任务在同一瞬间失败并立即重试时，你会发射一个同步爆发，比原始运行更猛烈地打击目标。添加指数退避加上随机抖动以便重试分散，限制尝试次数，并在失败时轮换 IP 而不是重用烧毁的那个。

```python
import random
from httpx import HTTPError

async def robust_fetch(client, url, tries=3):
    for attempt in range(tries):
        try:
            r = await client.get(url, timeout=httpx.Timeout(20.0))
            if r.status_code < 400 and looks_real(r.text):
                return r
        except HTTPError:
            pass
        # exponential backoff + jitter before the next attempt
        await asyncio.sleep((2 ** attempt) + random.uniform(0, 1))
    return None
```

![两个库的代码并列：httpx 在 AsyncClient 上一次性设置代理，并启用 http2 和 httpx.Timeout(20.0)；aiohttp 则在每次 session.get() 时传入 proxy=，并配合 ClientTimeout(total=30, connect=10) 和 TCPConnector(limit=20, limit_per_host=8)。卡片总结了两种代理约定。](https://res.cloudinary.com/dmqliab0t/image/upload/f_auto,q_auto,w_1200/v1785335149/blog/generated/qp-b9-010-diag2.png)
*两个库，两种代理约定：httpx 在客户端上设置代理，aiohttp 每请求设置代理。*

## 200 并不代表成功

最微妙的异步抓取错误是将 HTTP 200 视为完成。反爬虫系统返回 200 带有 CAPTCHA 页面、访问拒绝通知、空结果集或 JS 挑战——所以仅凭状态码判断“成功”的代理实际上在悄悄地给你提供被阻止的页面。验证*内容*：在信任响应之前检查已知元素、最小长度或挑战标记的缺失。这就是上面重试循环中的 `looks_real()` 门。

## 何时停止手动构建堆栈

上述模式——异步客户端、信号量、超时、轮换、内容验证——可以干净地处理大多数目标。但一旦网站层叠了 Cloudflare、TLS 指纹或重度客户端渲染，原始异步 HTTP 就开始失效，无论你的代理如何，因为 Python 的 TLS 握手看起来与 Chrome 的完全不同。在那条线上，一个[Scraper API](https://quantumproxies.io/scraper-api)，携带真实的浏览器指纹，按需轮换 IP 和渲染 JavaScript，比手动维护所有这些代码更少，成功率更高。我们关于[无头浏览器与 HTTP 成本](https://quantumproxies.io/blog/headless-browser-vs-http-requests-cost)的帖子涵盖了这种升级何时值得。

## 常见问题解答

### 如何在 aiohttp 中使用代理？

每请求传递代理 URL：`session.get(url, proxy="http://user:pass@host:port")`。与 `requests` 不同，aiohttp 不在会话上接受代理字典。始终覆盖默认的 User-Agent（`Python/3.x aiohttp/3.x` 是一个明显的机器人信号）并设置 `ClientTimeout`，以便死锁不能挂起协程。

### httpx 或 aiohttp 哪个更适合异步抓取？

默认选择 `httpx`：一个客户端同时支持同步和异步，内置 HTTP/2，代理是一个单一参数。当你想要最大并发控制时选择 `aiohttp`——显式连接池限制和每请求代理适合大型快速抓取。两者都很好；代理策略比库更重要。

### 我应该运行多少个并发请求？

不是你的机器允许的数量——而是目标和你的池能容忍的数量。以 10–20 的信号量开始，观察阻塞和错误率，并且只有在成功率保持高时才提高。在一个粘性 IP 上，保持在个位数。更广泛的 IP 轮换让你可以安全地运行更高的总并发。

### 为什么我的异步抓取器被阻止，而同步的没有？

因为并发集中信号：来自一个 IP 的许多同时请求是经典的机器人模式。通过轮换池分散负载，用信号量节流，添加抖动退避在重试时，并验证响应内容——200 仍然可能是一个挑战页面。没有轮换的速度就是导致你被阻止的原因。

这就是完整的异步模式：选择一个客户端，为其正确设置代理，用信号量限制并发，限制每个超时阶段，在逻辑边界轮换，并且永远不要信任裸 200。首先正确设置代理层，大多数阻止列表在你遇到之前就会消失。

[试用 QuantumProxies Scraper API](https://quantumproxies.io/scraper-api)