使用 Node.js 和网页抓取 API 构建可扩展的数据管道

Q: Node.js 抓取器应该发起多少并发请求?

你的抓取 API 套餐所允许的最大数量,不能更多。使用一个固定数量、设置为该限额的工作协程池,并把 429 响应当作应该放慢速度的信号。

一个通过网页抓取 API 获取数千个页面的 Node.js 管道：有限并发、重试机制、可恢复输出和经过测试的设计。

获取一个网页用网页抓取 API 只需要一行代码。但每晚抓取五万个页面,既不超出套餐限额,又不在进程崩溃时丢失已完成的工作,也不悄悄地漏掉失败的请求,这就是一个工程问题了。Node.js 很适合做这件事:这类工作几乎都是在等待网络响应,而这正是 Node 的事件循环擅长的地方。

本指南基于 Shifter 的网页抓取 API,用 Node.js 构建一个小型的、具备生产环境形态的管道:有限并发、针对正确错误类型的重试、可恢复的输出,以及独立的错误日志。整个实现大约六十行代码,不依赖任何第三方库,我们已经针对该 API 的一个模拟环境进行了测试。如果你还没发出过第一个请求,可以先阅读我们的网页抓取 API 新手指南。

管道的结构

一个抓取管道有四项任务:

一个队列,存放待抓取的 URL。
若干工作协程(workers),从队列中取出 URL 并调用 API,每次只处理固定数量。
错误处理,对可能稍后成功的情况进行重试,并记录无法成功的情况。
一种输出方式,能在崩溃后存活下来,并让重跑时跳过已完成的工作。

API 负责处理代理、浏览器渲染以及它那一侧的部分重试工作。管道的任务是以合适的速率向其输送请求,并保留一份可靠的执行记录。

并发:使用工作协程,而不是对所有任务一股脑地 `Promise.all`

最常见的 Node.js 错误做法,是把每个 URL 映射成一个请求,然后对全部请求 await Promise.all(...)。如果有五万个 URL,这会一次性发出五万个请求。API 会对其中大多数返回 429 Too Many Requests,因为每个套餐都有并发限制(在 Shifter 上,根据套餐不同,并发数从 20 到 500 个请求不等)。

解决方法是使用一个共享同一队列的固定数量的工作协程池。每个工作协程取出下一个 URL,处理它,然后再取下一个,这样任意时刻都恰好有 concurrency 个请求在处理中。不需要任何库,一个共享数组加一个循环即可实现。

重试:只针对稍后可能成功的错误

不是每个错误都值得重试。超时、渲染失败、429 或服务器错误在下一次尝试时可能会成功。而错误的 API 密钥或格式不正确的请求永远不会成功,重试只会浪费时间。该管道会对 408、422、429 和 500 进行重试,每次重试的等待时间会逐渐增加,并立即将其他所有错误记录为失败。我们关于网页抓取器的重试与退避策略的指南解释了为什么等待时间应当逐渐增加。

输出:只追加、可恢复

每个成功的结果一到达就会被追加写入一个以换行分隔的 JSON(NDJSON)文件:每行一个 URL,包含数据和时间戳。这样做有三个好处:

一次崩溃最多只会丢失正在处理中的请求。 已经写入的内容都会保留。
重跑成本很低。 启动时,管道会读取输出文件并跳过已经存在的 URL,这样你可以重启一次失败的运行,而不必为同一批页面重复付费。
失败记录会单独保存在一个 .errors 文件中,你可以查看它,或将其反馈到下一次运行中。

让重跑产生相同结果的设计思路,正是幂等抓取器这一核心理念,该文对此有更深入的介绍。

代码

import { appendFile, readFile } from 'node:fs/promises';

const API = process.env.SCRAPER_API_URL ?? 'https://scrape.shifter.io/v1';
const RETRY = new Set([408, 422, 429, 500]);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// Fetch one URL through the Web Scraping API, retrying only errors that can succeed later.
export async function fetchPage(url, { extractRules, attempts = 3, backoffMs = 2000 } = {}) {
  const params = new URLSearchParams({ api_key: process.env.SHIFTER_SCRAPER_KEY, url });
  if (extractRules) params.set('extract_rules', JSON.stringify(extractRules));
  for (let attempt = 1; ; attempt++) {
    const res = await fetch(`${API}?${params}`, { signal: AbortSignal.timeout(120_000) });
    if (res.ok) return extractRules ? res.json() : res.text();
    if (!RETRY.has(res.status) || attempt === attempts) {
      throw new Error(`HTTP ${res.status} for ${url}`);
    }
    await sleep(backoffMs * attempt);
  }
}

// URLs already written to the output file, so a restarted run skips them.
async function alreadyDone(outFile) {
  try {
    const lines = (await readFile(outFile, 'utf8')).split('\n').filter(Boolean);
    return new Set(lines.map((line) => JSON.parse(line).url));
  } catch {
    return new Set();
  }
}

// Run `concurrency` workers over the queue; each result becomes one NDJSON line.
export async function runPipeline(urls, { outFile, concurrency = 10, extractRules, backoffMs } = {}) {
  const done = await alreadyDone(outFile);
  const queue = urls.filter((u) => !done.has(u));
  const stats = { skipped: urls.length - queue.length, ok: 0, failed: 0 };

  async function worker() {
    for (let url = queue.shift(); url !== undefined; url = queue.shift()) {
      try {
        const data = await fetchPage(url, { extractRules, backoffMs });
        await appendFile(outFile, JSON.stringify({ url, data, at: new Date().toISOString() }) + '\n');
        stats.ok++;
      } catch (err) {
        await appendFile(`${outFile}.errors`, JSON.stringify({ url, error: err.message }) + '\n');
        stats.failed++;
      }
    }
  }

  await Promise.all(Array.from({ length: concurrency }, worker));
  return stats;
}

它只使用了 Node 内置的 fetch、AbortSignal.timeout 和 fs/promises,因此无需安装任何依赖,即可在任何当前的 Node.js 版本上运行。

将其运行在一批产品页面上,并让 API 返回 JSON 而非 HTML:

import { readFile } from 'node:fs/promises';
import { runPipeline } from './pipeline.mjs';

const urls = (await readFile('urls.txt', 'utf8')).split('\n').filter(Boolean);
const stats = await runPipeline(urls, {
  outFile: 'products.ndjson',
  concurrency: 20,                       // stay under your plan's concurrency limit
  extractRules: {
    name: { selector: 'h1', output: 'text' },
    price: { selector: '.price', output: 'text' },
  },
});
console.log(stats);                      // { skipped, ok, failed }

我们是如何测试的

我们让该管道针对一个本地的 API 替代实现运行,该实现提供了 22 个 URL:二十个正常页面、一个在成功之前连续两次返回 429 的页面,以及一个始终返回 401 的页面。测试检查了以下几点:

21 个页面成功,一个失败被记录在错误文件中;
那个不稳定的页面被重试,并在第三次尝试时成功,而 401 的页面只被尝试了一次;
在 concurrency: 5 的设置下,任何时刻处理中的请求都不超过五个;
第二次运行跳过了全部 21 个已完成的页面,只重试了那个失败项。

进一步扩展

上述单进程设计可以轻松应对每次运行数万个页面的规模。超出这个规模后:

把队列移出内存。 使用数据库表或消息队列,可以让多个进程共享工作,并在整台机器重启后依然存活。
结果一到达就加载进数据库。 NDJSON 是一种不错的落地格式;我们关于将网页抓取 API 数据导入 SQL的指南介绍了加载这一步。
应用背压(backpressure)。 如果抓取之后的步骤比抓取本身慢,应该让它拖慢工作协程的速度,而不是让结果堆积在内存中。参见分布式爬虫中的背压与流量控制。
关注数据指标。 跟踪每次运行的成功数、失败数和重试次数,并在失败率发生变化时发出警报。我们关于监控抓取管道的指南列出了重要的指标。

常见问题

Node.js 适合做网页抓取管道吗?

适合,前提是基于 API 构建。这类工作大部分时间都在等待网络响应,而 Node.js 能在单个进程中以较低开销处理大量并发请求。占用 CPU 较多的解析工作,最好交给 API 的提取规则处理,或者放到单独的步骤中完成。

Node.js 抓取器应该发起多少并发请求?

你的抓取 API 套餐所允许的最大数量,不能更多。使用一个固定数量、设置为该限额的工作协程池,并把 429 响应当作应该放慢速度的信号。

如何让抓取任务可恢复?

每个结果一到达就写入一个以 URL 为键的只追加文件或表,启动时跳过已经有结果的 URL。这样一次崩溃只会损失正在处理中的请求。

我需要一个队列库吗?

对于单进程来说不需要。一个共享数组配合固定数量的异步工作协程就足够了。当需要多个进程或多台机器共享工作时,再转向数据库或消息队列。

结论

一个可扩展的 Node.js 抓取管道需要四样东西:一个按套餐并发限额设定大小的固定工作协程池、只针对稍后可能成功的错误进行的重试、可供重跑恢复的只追加输出,以及一份独立的失败记录。上面的代码用大约六十行实现了这四点。API 负责处理代理和浏览器,而你的管道的任务是稳定地向它输送请求,并始终清楚地记录发生了什么。