异步 Python 网络 scraping：使用 aiohttp 和 Scrapeless 扩展到 10,000+ 个 URL

主要结论：

• 异步在 I/O 密集型抓取中击败同步大约 10–100 倍。 asyncio 的事件循环允许一个 Python 线程同时处理数百个正在进行的 HTTP 请求；而同步版则在每次套接字读取时阻塞，且每个 URL 支付完整的延迟。
• aiohttp 是标准的异步 HTTP 客户端。 单个 aiohttp.ClientSession 持有连接池、保持活动、cookie 和超时 — 将其与 asyncio.gather 配对以实现分发，并使用 asyncio.Semaphore 限制每个主机的请求数量。
• Scrapeless 住宅代理路由异步获取。 一个代理 URL 直接插入 aiohttp.ClientSession(... proxy=...)，为每个请求提供不同的住宅 IP，并通过嵌入在用户名中的国家代码固定出网地理位置。
• Scrapeless 抓取浏览器处理 JS 渲染的页面。 aiohttp 返回作为 JS 应用程序外壳（Next.js、React、Vue）的页面会被提升到云浏览器会话 — 通过 Scrapeless Python SDK 加上 Playwright 的异步 API 从异步 Python 连接。
• 故障不干扰其他操作。 asyncio.gather(return_exceptions=True) 防止单一的坏 URL 取消其他请求；失败的 URL 被送入死信列表进行单独审核，而不是进入内联循环。
• 免费开始。 新的 Scrapeless 账户包括免费抓取浏览器运行时 — 在 Scrapeless 注册。

介绍：为什么使用异步，串行抓取的成本

使用 requests 的同步 Python 抓取程序在每次套接字读取时会阻塞线程。抓取 1,000 个产品页面，每个请求 500 毫秒，实际所需时间大约是 500 秒 — 延迟全额支付，一次处理一个 URL。

asyncio 改变了这一点。事件循环在套接字传输中时让出控制权，允许下一个协程启动自己的请求，并在线程上将数百个抓取编织在一起。这同样的 1,000 个页面 — 限制每个主机 10 个并发请求 — 大约在 50 秒内完成。硬件相同，Python 相同，数据相同。

问题是：异步抓取有两种同步版本从未遇到过的失败模式。管道失败，在一个协程内抛出的异常可能会取消整个 gather，以及 目标侧压力，如果代理层无法在 IP 之间分散出网流量，紧张的在途池看起来与攻击无异。

本指南将讨论这两者。HTTP 层使用 aiohttp 和 Scrapeless 住宅代理，覆盖 195 个国家。JS 渲染层则提升至 Scrapeless 抓取浏览器，通过 Scrapeless Python SDK 和 Playwright 异步 API 从异步 Python 连接。

您可以用它做什么

• 大规模爬取静态目录。 书籍、文章、网站地图，任何提供渲染 HTML 的内容 — 异步分发将数小时的爬取变成数分钟的爬取。
• 运行并发数据拉取。 RSS、JSON API、网站地图索引；在数百个端点上进行分发，限制并发访问。
• 跨地区价格监测。 将 Scrapeless 出网流量固定在美国、英国、德国、日本，并在多个地区并行抓取相同的产品页面。
• 审计抓取您的网站。 异步在几分钟内扫过 10,000 个 URL 的网站地图，而不是几个小时，并报告死链接和慢路径。
• 为下游管道提供数据。 异步层将渲染的 HTML 或 JSON 直接传入 Postgres、Snowflake 或 Kafka，而不会出现线程池瓶颈。
• 选择性提升。 对于约 70% 的页面（发送渲染标记），使用 aiohttp 执行便宜的 HTTP；仅对 JS 密集型的少数页面启动云浏览器会话。

在 Scrapeless，我们仅访问公开可用的数据，同时严格遵守适用的法律、法规和网站隐私政策。本帖中的内容仅用于演示目的。

为什么选择 Scrapeless 用于异步抓取

Scrapeless 抓取浏览器是一个可定制的反检测云浏览器，专为网络爬虫和 AI 代理设计；Scrapeless 住宅代理则是其底层的代理层。具体针对异步 Python 管道，这个组合提供：

• 覆盖 195 个国家的住宅代理，以单个 HTTP 代理 URL 形式暴露，可直接放入 aiohttp.ClientSession(... proxy=...)。
• 每个请求的地理固定，通过嵌入代理凭证的国家代码 — 无需每个请求的握手成本，也无需每个协程的会话重建。
• 粘性会话选项，对于需要在多步骤登录或分页遍历中保持相同 IP 的流程，以及对其他所有流量使用轮换 IP。
• 云端 JS 渲染，当页面重度使用 React/Vue/Next.js 时 — Python SDK 会生成一个 browser_ws_endpoint，您可以通过 Playwright 的异步 API 连接。
• 两个层次共用一个 API 密钥 — 代理和抓取浏览器通过相同的 Scrapeless 账户计费。

在 Scrapeless 免费计划中获取您的 API 密钥。

先决条件

• Python 3.10 或更高版本
• 一个 Scrapeless 账户和 API 密钥 — 在 app.scrapeless.com 注册
• 熟悉 async/await 和事件循环模型
• 一个终端

步骤 1 — 安装 asyncio、aiohttp 和 Scrapeless SDK

aiohttp 自带内建的 asyncio 支持。scrapeless SDK 为第 6 步的升级级别生成云浏览器会话。Playwright 的异步 API 是驱动 Scrapeless Scraping Browser 的标准异步 Python 方式：

pip install aiohttp scrapeless playwright
playwright install chromium

playwright install chromium 下载一个本地 CDP 客户端一次；实际渲染仍然在 Scrapeless 的云中进行 — 本地 Chromium 只是协议发言者。

步骤 2 — 配置你的 Scrapeless 凭据

将你的 Scrapeless API 密钥、频道 ID 和住宅代理频道密码导出为环境变量。所有三者在 Scrapeless 仪表盘的 Proxies → Residential 中可见，网址为 app.scrapeless.com — 点击 Generate，仪表盘会打印出格式为 <GATEWAY>:<PORT>:<CHANNEL_ID>-proxy-country_US-r_10m-s_<SESSION_ID>:<PASSWORD> 的以冒号分隔的字符串：

export SCRAPELESS_API_KEY="your_api_token_here"
export SCRAPELESS_CHANNEL_ID="your_channel_id"          # 在用户名开头打印出来
export SCRAPELESS_PROXY_PASS="your_channel_password"
export SCRAPELESS_PROXY_GATEWAY="gw-us.scrapeless.io"   # 请参见下面的区域网关

区域网关： gw-us.scrapeless.io (美洲), gw-eu.scrapeless.io (欧洲), gw-ap.scrapeless.io (亚太地区)。选择离你运行时最近的网关以保持握手延迟低；无论你通过哪个网关连接，出口国家仍由 country_<CC> 用户名参数控制。所有的端口都是 8789。

住宅代理用户名由四个参数构成：

• <CHANNEL_ID> — 你的频道标识符（在仪表盘用户名开头打印）。
• country_<CC> — 以两字母形式表示的国家代码。Scrapeless 使用 country_US、country_UK、country_DE、country_JP 等（注意：使用 UK，而不是 ISO 的 GB）。
• r_<duration> — sticky-session 轮换间隔（例如 r_10m 在旋转之前保持相同的 IP 10 分钟）。
• s_<SESSION_ID> — sticky-session 标识符；在请求之间重用相同的 s_<id> 以在持续时间窗口内保留相同的 IP。

去掉 r_ 和 s_ 可以获取旋转的 IP（每个请求一个全新的住宅 IP）。对于需要会话连续性的流，例如在登录后的分页遍历，保留它们。

步骤 3 — 基础：使用 aiohttp + Scrapeless 代理进行单次异步抓取

最小的功能性异步爬虫。一个 ClientSession，一个 GET，一个通过住宅代理返回的 HTML 负载：

import asyncio
import os
import aiohttp

PROXY = (
    f"http://{os.environ['SCRAPELESS_CHANNEL_ID']}-proxy-country_US"
    f":{os.environ['SCRAPELESS_PROXY_PASS']}"
    f"@{os.environ['SCRAPELESS_PROXY_GATEWAY']}:8789"
)

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    timeout = aiohttp.ClientTimeout(total=30)
    async with session.get(url, proxy=PROXY, timeout=timeout) as resp:
        resp.raise_for_status()
        return await resp.text()

async def main() -> None:
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, "https://books.toscrape.com/")
        print(f"通过美国住宅出口抓取了 {len(html):,} 个字符")

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

这个代码片段早期锁定了三件事：

• ClientSession 只创建一次并重复使用。 每次 session.get(...) 共享相同的连接池 — 每请求重新创建会话会破坏异步的目的。
• 代理 URL 是按请求传递的，而不是按会话。 这使得相同的 ClientSession 可以自由路由不同的请求穿过不同的国家。
• ClientTimeout(total=30) 限制每个请求。单个挂起的连接不会阻塞其余的 gather。

步骤 4 — 高级：使用 asyncio.gather 和 Semaphore 上限进行并发抓取

没有并发上限地扩展到 100 个 URL 是爬虫在 10 秒内被阻止的方式。经典模式是 asyncio.Semaphore 来限制每个主机的在飞请求：

import asyncio
import os
import aiohttp

PROXY = (
    f"http://{os.environ['SCRAPELESS_CHANNEL_ID']}-proxy-country_US"
    f":{os.environ['SCRAPELESS_PROXY_PASS']}"
    f"@{os.environ['SCRAPELESS_PROXY_GATEWAY']}:8789"
)

# 设定每个主机最大 5 个并发请求。根据目标进行调整 — 公共目录
# 可以忍受更高的并发，而防爬虫保护的源希望更低。
PER_HOST = asyncio.Semaphore(5)

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    async with PER_HOST:
        timeout = aiohttp.ClientTimeout(total=30)
        async with session.get(url, proxy=PROXY, timeout=timeout) as resp:

resp.raise_for_status()
            return await resp.text()

async def main() -> None:
    urls = [
        f"https://books.toscrape.com/catalogue/page-{n}.html"
        for n in range(1, 51)  # 50 个目录页面
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)
    print(f"获取了 {len(pages)} 个页面，总共 {sum(len(p) for p in pages):,} 个字符")

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

asyncio.Semaphore(5) 是承重线。如果没有它，asyncio.gather 同时启动所有 50 个协程，而网关可能会限制或拒绝其中一半。使用它时，最多只有 5 个正在进行；其余的在事件循环中等待，直到有槽位释放。

对于多主机的分发，从每个源创建一个信号量，并根据主机名为其键入——这样一个源的停滞不会对其他源的抓取造成阻塞。

获取您的 API 密钥：

Scrapeless 免费计划

第 5 步 - 处理失败而不阻塞管道

在协程内的一个 raise_for_status() 将取消整个 gather 并丢失每一个正在进行的结果。两种防御措施：

防御措施 1: return_exceptions=True. 告诉 gather 捕获异常作为值，而不是传播它们。无论如何，管道都能完成，调用者之后决定对哪些 URL 采取行动。

防御措施 2: 死信列表。 将失败的 URL 收集到一个单独的结构中以供单独查看。失败处理保持在带外，即当成功和失败路径不交错时，异步管道保持清晰。

import asyncio
import json
import aiohttp

async def fetch_safe(session, url):
    try:
        async with session.get(
            url, timeout=aiohttp.ClientTimeout(total=30)
        ) as resp:
            resp.raise_for_status()
            return {"url": url, "html": await resp.text()}
    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
        return {"url": url, "error": repr(exc)}

async def main(urls):
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *(fetch_safe(session, u) for u in urls)
        )

    ok = [r for r in results if "html" in r]
    failed = [r for r in results if "error" in r]
    print(f"成功: {len(ok)}   失败: {len(failed)}")

    # 死信文件以供单独查看 - 管道在失败时永不阻塞
    with open("dead_letter.jsonl", "w", encoding="utf-8") as f:
        for r in failed:
            f.write(json.dumps(r) + "\n")

需要注意两点：

• 错误包裹（{"url": ..., "error": ...}）与成功包裹的形状相同，只是键不同。下游消费者通过检查哪个键存在来分支，而无需解析异常文本。
• aiohttp.ClientError 覆盖了常见的故障情况（连接中断、格式错误的响应、DNS 问题）；asyncio.TimeoutError 是由 ClientTimeout 引发的。捕获这两者覆盖了 ~95% 的真实世界异步抓取。

这段代码故意 不 做的事情：成功路径中的任何内容不会重新请求失败的 URL。死信处理属于一个单独的运行——使用不同的代理国家、不同的并发上限或第 6 步的云浏览器等级。将其混合在一起将一个异步抓取器变成两个交错的控制流，而错误最终会出现在交错中。

第 6 步 - 将 JS 渲染的页面升级到 Scrapeless 抓取浏览器

aiohttp 返回原始发送的字节。对于 Next.js、React 和 Vue 应用程序，这些字节是一个空的 <div id="root"> 加一个脚本标签——实际内容是在客户端呈现的。普通 HTTP 无法渲染这一点；而云浏览器可以。

最干净的升级模式：对约 70% 的页面使用 aiohttp 发送已渲染的 HTML，升级对 JS 渲染的少数页面到 Scrapeless 抓取浏览器。Python SDK 创建一个云浏览器会话，并提供一个 browser_ws_endpoint；Playwright 的异步 API 通过 Chrome DevTools 协议连接到它：

import asyncio
from scrapeless import Scrapeless
from scrapeless.types import ICreateBrowser
from playwright.async_api import async_playwright

async def render_via_cloud_browser(url: str, country: str = "US") -> str:
    client = Scrapeless()  # 从环境中读取 SCRAPELESS_API_KEY
    session = client.browser.create(
        ICreateBrowser(proxy_country=country, session_ttl=240)
    )

    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(session.browser_ws_endpoint)
        context = browser.contexts[0] if browser.contexts else await browser.new_context()
        page = context.pages[0] if context.pages else await context.new_page()
        await page.goto(url, wait_until="networkidle", timeout=60_000)
        html = await page.content()
        await browser.close()
        return html

async def main():
    # quotes.toscrape.com/js/ 是标准的“需要 JS”的沙盒。
    # 普通 HTTP 返回 0 个引用元素；云渲染返回 10 个。
    html = await render_via_cloud_browser("https://quotes.toscrape.com/js/")
    print(f"渲染了 {len(html):,} 个字符，包括绘制后的 DOM")

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

session.browser_ws_endpoint 是一个 wss://browser.scrapeless.com/...?token=... URL。Playwright 的 connect_over_cdp 会对该端点进行 CDP 通信；渲染在 Scrapeless 的云中进行，而不是在本地机器上。本地的 playwright install chromium 步骤只是用于协议客户端。

session_ttl=240 可以让会话保持活跃 4 分钟——足以在单个页面上进行多步骤的遍历。对于长时间运行的网络爬虫，每个 URL 或逻辑工作单元生成一个新的会话；云会话创建成本较低。

第 7 步——将所有内容组合起来：分层异步爬虫

异步爬虫管道的现实形状是 HTTP 优先，浏览器其次：尝试 aiohttp，当响应为空或被阻塞时升级至 Scrapeless 爬虫。两个层次之间共享并发限制，但位于不同的信号量中——云浏览器会话比 HTTP 请求稀缺。

import asyncio
import os
import aiohttp
from scrapeless import Scrapeless
from scrapeless.types import ICreateBrowser
from playwright.async_api import async_playwright

PROXY = (
    f"http://{os.environ['SCRAPELESS_CHANNEL_ID']}-proxy-country_US"
    f":{os.environ['SCRAPELESS_PROXY_PASS']}"
    f"@{os.environ['SCRAPELESS_PROXY_GATEWAY']}:8789"
)
HTTP_LIMIT = asyncio.Semaphore(10)      # aiohttp 层
BROWSER_LIMIT = asyncio.Semaphore(3)    # 云浏览器层

async def http_fetch(session: aiohttp.ClientSession, url: str) -> str | None:
    async with HTTP_LIMIT:
        try:
            async with session.get(
                url, proxy=PROXY,
                timeout=aiohttp.ClientTimeout(total=30),
            ) as resp:
                resp.raise_for_status()
                return await resp.text()
        except (aiohttp.ClientError, asyncio.TimeoutError):
            return None

async def browser_fetch(client: Scrapeless, url: str) -> str:
    async with BROWSER_LIMIT:
        session = client.browser.create(
            ICreateBrowser(proxy_country="US", session_ttl=240)
        )
        async with async_playwright() as p:
            browser = await p.chromium.connect_over_cdp(session.browser_ws_endpoint)
            context = (
                browser.contexts[0] if browser.contexts
                else await browser.new_context()
            )
            page = await context.new_page()
            await page.goto(url, wait_until="networkidle", timeout=60_000)
            html = await page.content()
            await browser.close()
            return html

from urllib.parse import urlparse

# (a) 知名的 JS 重度托管总是升级——最可靠的信号。
JS_HEAVY_HOSTS = {"quotes.toscrape.com"}

def should_escalate(url: str, html: str | None) -> bool:
    # (a) 白名单命中——显式的 JS 重度托管。
    if urlparse(url).hostname in JS_HEAVY_HOSTS:
        return True
    # (b) 解析后信号——空主体或可识别的应用外壳。
    if html is None or len(html) < 2000 or '<div id="root"></div>' in html:
        return True
    return False

async def scrape_one(http_session, client, url):
    html = await http_fetch(http_session, url)
    tier = "http"
    if should_escalate(url, html):
        tier = "browser"
        html = await browser_fetch(client, url)
    return {"url": url, "tier": tier, "html_len": len(html) if html else 0}

async def main(urls):
    client = Scrapeless()
    async with aiohttp.ClientSession() as http_session:
        results = await asyncio.gather(
            *(scrape_one(http_session, client, u) for u in urls)
        )
    return results

if __name__ == "__main__":
    urls = [
        "https://books.toscrape.com/",       # 静态 — aiohttp 层
        "https://quotes.toscrape.com/js/",   # JS — 升级
    ]
    print(asyncio.run(main(urls)))

should_escalate 结合了文中提到的两个信号：（a）一个“已知的 JS 重度”托管的显式白名单，以及（b）一个解析后信号（空主体 / 应用外壳）。白名单是更可靠的杠杆——即便主体为空，Next.js 或 React 外壳通常也会超过 2,000 字节的阈值，因此仅通过 <div id="root"></div> 的检查可能会失效。主机名检查在任何字节计数之前触发。

您收获的内容

该管道会产生如下格式的字典列表：

[
  {
    "url": "https://books.toscrape.com/",
    "tier": "http",
    "html_len": 51274
  },
  {
    "url": "https://quotes.toscrape.com/js/",
    "tier": "browser",
    "html_len": 9246
  }
]

运行此模式的诚实观察：

• 冷连接成本是现实的。 在新的 ClientSession 上的第一次请求会支付 TLS + DNS；在同一个会话上的后续请求会重用连接。不要根据每个请求重新创建会话。
• 并发上限取决于目标，而不是 aiohttp。 对于公共目录，每个主机五个并发请求是一个安全的起点；对于受防爬虫保护的源，三个是更安全的；对于友好的 API 十个是现实的。
• 云浏览器会话的生命周期超过单页加载。 如果一个流程需要登录、遍历和提取，那么为每个逻辑工作单元创建一个会话，并在该单元内的页面之间重用它 — 在同一个会话中 context.new_page() 是便宜的。
• DNS 解析在 aiohttp 内部进行。 连接器会缓存解析的 IP，直到 ClientSession 结束。对于长时间运行的爬虫，每几小时回收一次会话，以防 DNS 失效。
• ClientTimeout(total=30) 是每个请求，而不是每个 gather。 1,000 个 URL 的分发不会在 30 秒内超时 — 每个请求都有自己的 30 秒预算。

结论：扩展你的异步 Python 爬虫

异步模式归结为四个步骤：启动一个 ClientSession，通过 Semaphore 限制并发，通过 Scrapeless 住宅代理路由分发，并通过 Python SDK 和 Playwright 的异步 API 将渲染的少数内容升级到 Scrapeless Scraping Browser。

要深入了解路由每个异步抓取的住宅代理层，请参阅 什么是 SSL 代理？。

在代理用户名中固定出境国家后缀，保持每主机 Semaphore 的小范围，根据成功与失败的包络形状进行分支，而不是在内部捕获异常，并将空 HTTP 响应视为升级的信号 — 而不是答案。

准备构建你的 AI 驱动的数据管道吗？

加入我们的社区，领取免费计划，与构建异步抓取管道的开发人员连接：Discord · Telegram。

在 Scrapeless 注册，获取免费的 Scraping Browser 运行时，并将上述模式调整为管道所需的目录、数据流和地区。定价详情请见 scrapeless.com/en/pricing；住宅代理的文档请见 scrapeless.com/en/product/proxy-solutions；完整的 SDK 参考请见 docs.scrapeless.com。

常见问题

Q1：每个主机我应该运行多少个并发请求？

对于没有反爬虫堆栈的公共目录，每个主机10个并发请求是一个安全的上限。对于受防爬虫保护的源，三个更为现实。Semaphore 是杠杆；开始时设定较低的值，观察 429 响应，然后进行调优。

Q2：如果我的目标没有被我的数据中心阻止，我是否需要 Scrapeless 住宅代理？

对于没有被阻止的 HTTP 目标，不需要 — aiohttp 无需代理即可工作。当目标地理限制（你需要美国/英国/日本出境）、数据中心 IP 被限速或阻止，或者你需要每个请求一个新的住宅 IP 在多个源之间分散在途池时，Scrapeless 代理层才能发挥作用。

Q3：我何时应该从 aiohttp 升级到 Scrapeless Scraping Browser？

当 aiohttp 返回的 HTML 是一个没有内容的 JS 应用程序壳时。启发式方法：在第一次获取后计算你关心的元素数量；如果计数为零或远低于预期，该页面在客户端渲染。云浏览器层处理这些情况。

Q4：异步抓取合法吗？

异步是一种传输模式；合法性取决于你抓取什么、从哪里以及在什么条件下。公开可见的数据一般是可以访问的；不同的法域有所不同；网站服务条款适用；请咨询法律顾问以获取高风险用例的建议。Scrapeless 仅访问公开可用的数据。

Q5：我可以在没有 Scrapeless Scraping Browser 的情况下使用 aiohttp 吗？

可以。aiohttp 层（步骤 3-5）可以作为任何发送渲染 HTML 的目标的完整异步爬虫。Scrapeless Scraping Browser 是升级层 — 仅在 HTTP 层返回空结果时调用。

Q6：我如何将出境固定到特定国家？

国家在 Scrapeless 住宅代理用户名中作为 country_<CC> （大写的两字母代码，用下划线分隔）：country_US、country_UK、country_DE、country_JP。替换用户名字符串中的这一部分，网关将每个请求路由通过该国家的住宅 IP。对于浏览器层的请求，在创建云浏览器会话时将 proxy_country="US" 传递给 ICreateBrowser(...)。

Q7：为什么使用 Playwright 的异步 API 而不是同步浏览器客户端？
同步浏览器客户端会阻塞事件循环。asyncio 的整个要点是保持循环自由；在协程内部调用同步的 page.goto(...) 会暂停所有其他进行中的任务。Playwright 的 async_playwright 是唯一一个保持与云浏览器层协程友好的标准 Python 选项。Scrapeless SDK 仍然生成会话——Playwright 仅通过 browser_ws_endpoint 与 CDP 进行通信。