跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

使用 Queues 和 Browser Run 构建 Web 爬虫

如何使用 Queues 和 Browser Run 驱动 Web 爬虫的示例。

最后更新 查看 MarkdownAgent 设置

本教程说明如何使用 Queues、Browser Run 和 Puppeteer 构建和部署 Web 爬虫。

Puppeteer 是一个用于自动化 Chrome/Chromium 浏览器交互的高级库。在每个提交的页面上，爬虫将查找指向 cloudflare.com 的链接数量并截取站点截图，将结果保存到 Workers KV。

您可以使用 Puppeteer 请求页面上的所有图像、保存站点使用的颜色等。

前提条件

  1. 注册 Cloudflare 账户 ↗。
  2. 安装 Node.js ↗。

Node.js 版本管理器

使用 Volta ↗ 或 nvm ↗ 等 Node 版本管理器，以避免权限问题并切换 Node.js 版本。本指南后续将介绍的 Wrangler 需要 Node 版本 16.17.0 或更高。

1. 创建新的 Workers 应用程序

要开始使用，请使用 create-cloudflare CLI ↗ 创建 Worker 应用程序。打开终端窗口并运行以下命令：

npm create cloudflare@latest -- queues-web-crawler

进行设置时，请选择以下选项：

  • 对于 What would you like to start with?，选择 Hello World example。
  • 对于 Which template would you like to use?，选择 Worker only。
  • 对于 Which language do you want to use?，选择 TypeScript。
  • 对于 Do you want to use git for version control?，选择 Yes。
  • 对于 Do you want to deploy your application?，选择 No（部署前我们还会做一些修改）。

然后，进入新创建的目录：

cd queues-web-crawler

2. 创建 KV 命名空间

我们需要创建 KV 存储。这可以通过 Cloudflare 仪表板或 Wrangler CLI 完成。在本教程中，我们将使用 Wrangler CLI。

npx wrangler kv namespace create crawler_links
npx wrangler kv namespace create crawler_screenshots
🌀 Creating namespace with title "web-crawler-crawler-links"
✨ Success!
Add the following to your configuration file in your kv_namespaces array:
[[kv_namespaces]]
binding = "crawler_links"
id = "<GENERATED_NAMESPACE_ID>"

🌀 Creating namespace with title "web-crawler-crawler-screenshots"
✨ Success!
Add the following to your configuration file in your kv_namespaces array:
[[kv_namespaces]]
binding = "crawler_screenshots"
id = "<GENERATED_NAMESPACE_ID>"

将 KV 绑定添加到 Wrangler 配置文件

然后，在 Wrangler 文件中添加终端生成的值：

{
	"kv_namespaces": [
		{
			"binding": "CRAWLER_SCREENSHOTS_KV",
			"id": "<GENERATED_NAMESPACE_ID>",
		},
		{
			"binding": "CRAWLER_LINKS_KV",
			"id": "<GENERATED_NAMESPACE_ID>",
		},
	],
}
[[kv_namespaces]]
binding = "CRAWLER_SCREENSHOTS_KV"
id = "<GENERATED_NAMESPACE_ID>"

[[kv_namespaces]]
binding = "CRAWLER_LINKS_KV"
id = "<GENERATED_NAMESPACE_ID>"

3. 设置 Browser Run

现在，您需要为 Worker 设置 Browser Run。

在当前目录中，安装 Cloudflare 的 Puppeteer 分支 以及 robots-parser ↗：

npm i -D @cloudflare/puppeteer
npm i robots-parser

然后，添加 Browser Run 绑定。添加 Browser Run 绑定使 Worker 能够访问您将使用 Puppeteer 控制的无头 Chromium 实例。

{
	"browser": {
		"binding": "CRAWLER_BROWSER",
	},
}
[browser]
binding = "CRAWLER_BROWSER"

4. 设置 Queue

现在，我们需要设置 Queue。

npx wrangler queues create queues-web-crawler
Outputtxt
Creating queue queues-web-crawler.
Created queue queues-web-crawler.

将 Queue 绑定添加到 Wrangler 配置

然后，在 Wrangler 文件中添加以下内容：

{
	"queues": {
		"consumers": [
			{
				"queue": "queues-web-crawler",
				"max_batch_timeout": 60,
			},
		],
		"producers": [
			{
				"queue": "queues-web-crawler",
				"binding": "CRAWLER_QUEUE",
			},
		],
	},
}
[[queues.consumers]]
queue = "queues-web-crawler"
max_batch_timeout = 60

[[queues.producers]]
queue = "queues-web-crawler"
binding = "CRAWLER_QUEUE"

在消费者队列中添加 max_batch_timeout 为 60 秒很重要，因为它允许 Queue 在更长时间内收集消息到批次中。这有助于管理 Browser Run 速率限制，并可通过单个浏览器实例在单个批次中处理多个 URL 来提高效率。

您的最终 Wrangler 文件应类似于下面的文件。

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "web-crawler",
	"main": "src/index.ts",
	// Set this to today's date
	"compatibility_date": "2026-08-17",
	"compatibility_flags": ["nodejs_compat"],
	"kv_namespaces": [
		{
			"binding": "CRAWLER_SCREENSHOTS_KV",
			"id": "<GENERATED_NAMESPACE_ID>",
		},
		{
			"binding": "CRAWLER_LINKS_KV",
			"id": "<GENERATED_NAMESPACE_ID>",
		},
	],
	"browser": {
		"binding": "CRAWLER_BROWSER",
	},
	"queues": {
		"consumers": [
			{
				"queue": "queues-web-crawler",
				"max_batch_timeout": 60,
			},
		],
		"producers": [
			{
				"queue": "queues-web-crawler",
				"binding": "CRAWLER_QUEUE",
			},
		],
	},
}
"$schema" = "./node_modules/wrangler/config-schema.json"
name = "web-crawler"
main = "src/index.ts"
# Set this to today's date
compatibility_date = "2026-08-17"
compatibility_flags = [ "nodejs_compat" ]

[[kv_namespaces]]
binding = "CRAWLER_SCREENSHOTS_KV"
id = "<GENERATED_NAMESPACE_ID>"

[[kv_namespaces]]
binding = "CRAWLER_LINKS_KV"
id = "<GENERATED_NAMESPACE_ID>"

[browser]
binding = "CRAWLER_BROWSER"

[[queues.consumers]]
queue = "queues-web-crawler"
max_batch_timeout = 60

[[queues.producers]]
queue = "queues-web-crawler"
binding = "CRAWLER_QUEUE"

5. 将绑定添加到环境

在 src/index.ts 中将绑定添加到环境接口，以便 TypeScript 正确类型化绑定。队列类型为 Queue<Message>，其中 Message 在下一步中定义。

import type { BrowserWorker } from "@cloudflare/puppeteer";

export interface Env {
	CRAWLER_QUEUE: Queue<Message>;
	CRAWLER_SCREENSHOTS_KV: KVNamespace;
	CRAWLER_LINKS_KV: KVNamespace;
	CRAWLER_BROWSER: BrowserWorker;
}

6. 提交要爬取的链接

向 Worker 添加 fetch() 处理程序以提交要爬取的链接。

type Message = {
	url: string;
};

export interface Env {
	CRAWLER_QUEUE: Queue<Message>;
	// ... etc.
}

export default {
	async fetch(req, env, ctx): Promise<Response> {
		await env.CRAWLER_QUEUE.send({ url: await req.text() });
		return new Response("Success!");
	},
} satisfies ExportedHandler<Env>;

这将接受任何子路径的请求并将请求正文转发以进行爬取。它期望请求正文仅包含 URL。在生产环境中，您应检查请求是否为 POST 请求且正文包含格式良好的 URL。为简单起见，此处已省略。

7. 使用 Puppeteer 爬取

向 Worker 添加 queue() 处理程序以处理您发送的链接。

import puppeteer from "@cloudflare/puppeteer";
import robotsParser from "robots-parser";

async queue(batch, env, ctx): Promise<void> {
  let browser: puppeteer.Browser | null = null;
  try {
    browser = await puppeteer.launch(env.CRAWLER_BROWSER);
  } catch {
    batch.retryAll();
	return;
  }

  for (const message of batch.messages) {
    const { url } = message.body;

    let isAllowed = true;
    try {
      const robotsTextPath = new URL(url).origin + "/robots.txt";
      const response = await fetch(robotsTextPath);

      const robots = robotsParser(robotsTextPath, await response.text());
      isAllowed = robots.isAllowed(url) ?? true; // respect robots.txt!
    } catch {}

    if (!isAllowed) {
      message.ack();
      continue;
    }

	// TODO: crawl!
    message.ack();
  }

  await browser.close();
},

这是爬虫的骨架。它启动 Puppeteer 浏览器并遍历 Queue 收到的消息。它获取站点的 robots.txt 并使用 robots-parser 检查该站点是否允许爬取。若不允许爬取，消息被 ack，从 Queue 中移除。若允许爬取，您可以继续爬取站点。

puppeteer.launch() 包装在 try...catch 中，以便在浏览器启动失败时重试整个批次。浏览器启动可能因超过每个账户的浏览器数量限制而失败。

type Result = {
	numCloudflareLinks: number;
	screenshot: ArrayBuffer;
};

const crawlPage = async (url: string): Promise<Result> => {
	const page = await (browser as puppeteer.Browser).newPage();

	await page.goto(url, {
		waitUntil: "load",
	});

	const numCloudflareLinks = await page.$$eval("a", (links) => {
		links = links.filter((link) => {
			try {
				return new URL(link.href).hostname.includes("cloudflare.com");
			} catch {
				return false;
			}
		});
		return links.length;
	});

	await page.setViewport({
		width: 1920,
		height: 1080,
		deviceScaleFactor: 1,
	});

	return {
		numCloudflareLinks,
		screenshot: ((await page.screenshot({ fullPage: true })) as Buffer).buffer,
	};
};

此辅助函数在 Puppeteer 中打开新页面并导航到提供的 URL。numCloudflareLinks 使用 Puppeteer 的 $$eval（等同于 document.querySelectorAll）查找指向 cloudflare.com 页面的链接数量。检查链接的 href 是否指向 cloudflare.com 页面包装在 try...catch 中，以处理 href 可能不是 URL 的情况。

然后，该函数设置浏览器视口大小并截取完整页面截图。截图以 Buffer 返回，以便转换为 ArrayBuffer 并写入 KV。

要启用递归爬取链接，在检查 Cloudflare 链接数量后添加代码片段，从队列消费者向队列本身递归发送消息。递归过深（爬取时可能发生）会导致 Durable Object Subrequest depth limit exceeded. 错误。若发生，会被捕获，但链接不会重试。

// const numCloudflareLinks = await page.$$eval("a", (links) => { ...

await page.$$eval("a", async (links) => {
	const urls: MessageSendRequest<Message>[] = links.map((link) => {
		return {
			body: {
				url: link.href,
			},
		};
	});
	try {
		await env.CRAWLER_QUEUE.sendBatch(urls);
	} catch {} // do nothing, likely hit subrequest limit
});

// await page.setViewport({ ...

然后，在 queue 处理程序中，对 URL 调用 crawlPage。

// in the `queue` handler:
// ...
if (!isAllowed) {
	message.ack();
	continue;
}

try {
	const { numCloudflareLinks, screenshot } = await crawlPage(url);
	const timestamp = new Date().getTime();
	const resultKey = `${encodeURIComponent(url)}-${timestamp}`;
	await env.CRAWLER_LINKS_KV.put(resultKey, numCloudflareLinks.toString(), {
		metadata: { date: timestamp },
	});
	await env.CRAWLER_SCREENSHOTS_KV.put(resultKey, screenshot, {
		metadata: { date: timestamp },
	});
	message.ack();
} catch {
	message.retry();
}

// ...

此代码片段将 crawlPage 的结果保存到相应的 KV 命名空间。若发生意外错误，URL 将被重试并再次发送到队列。

在 KV 中保存爬取时间戳有助于避免过于频繁地爬取。

在检查 robots.txt 之前添加代码片段，检查 KV 中是否在一小时内有爬取。这将列出以相同 URL 开头的所有 KV 键（同一页面的爬取），并检查是否在一小时内进行过任何爬取。若在一小时内进行过爬取，消息被 ack 且不重试。

type KeyMetadata = {
  date: number;
};

// in the `queue` handler:
// ...
for (const message of batch.messages) {
  const sameUrlCrawls = await env.CRAWLER_LINKS_KV.list({
    prefix: `${encodeURIComponent(url)}`,
  });

  let shouldSkip = false;
  for (const key of sameUrlCrawls.keys) {
    if (timestamp - (key.metadata as KeyMetadata)?.date < 60 * 60 * 1000) {
      // if crawled in last hour, skip
      message.ack();
      shouldSkip = true;
      break;
    }
  }
  if (shouldSkip) {
    continue;
  }

  let isAllowed = true;
  // ...

最终脚本如下。

import puppeteer, { BrowserWorker } from "@cloudflare/puppeteer";
import robotsParser from "robots-parser";

type Message = {
	url: string;
};

export interface Env {
	CRAWLER_QUEUE: Queue<Message>;
	CRAWLER_SCREENSHOTS_KV: KVNamespace;
	CRAWLER_LINKS_KV: KVNamespace;
	CRAWLER_BROWSER: BrowserWorker;
}

type Result = {
	numCloudflareLinks: number;
	screenshot: ArrayBuffer;
};

type KeyMetadata = {
	date: number;
};

export default {
	async fetch(req, env, ctx): Promise<Response> {
		// util endpoint for testing purposes
		await env.CRAWLER_QUEUE.send({ url: await req.text() });
		return new Response("Success!");
	},
	async queue(batch, env, ctx): Promise<void> {
		const crawlPage = async (url: string): Promise<Result> => {
			const page = await (browser as puppeteer.Browser).newPage();

			await page.goto(url, {
				waitUntil: "load",
			});

			const numCloudflareLinks = await page.$$eval("a", (links) => {
				links = links.filter((link) => {
					try {
						return new URL(link.href).hostname.includes("cloudflare.com");
					} catch {
						return false;
					}
				});
				return links.length;
			});

			// to crawl recursively - uncomment this!
			/*await page.$$eval("a", async (links) => {
        const urls: MessageSendRequest<Message>[] = links.map((link) => {
          return {
            body: {
              url: link.href,
            },
          };
        });
        try {
          await env.CRAWLER_QUEUE.sendBatch(urls);
        } catch {} // do nothing, might've hit subrequest limit
      });*/

			await page.setViewport({
				width: 1920,
				height: 1080,
				deviceScaleFactor: 1,
			});

			return {
				numCloudflareLinks,
				screenshot: ((await page.screenshot({ fullPage: true })) as Buffer)
					.buffer,
			};
		};

		let browser: puppeteer.Browser | null = null;
		try {
			browser = await puppeteer.launch(env.CRAWLER_BROWSER);
		} catch {
			batch.retryAll();
			return;
		}

		for (const message of batch.messages) {
			const { url } = message.body;
			const timestamp = new Date().getTime();
			const resultKey = `${encodeURIComponent(url)}-${timestamp}`;

			const sameUrlCrawls = await env.CRAWLER_LINKS_KV.list({
				prefix: `${encodeURIComponent(url)}`,
			});

			let shouldSkip = false;
			for (const key of sameUrlCrawls.keys) {
				if (timestamp - (key.metadata as KeyMetadata)?.date < 60 * 60 * 1000) {
					// if crawled in last hour, skip
					message.ack();
					shouldSkip = true;
					break;
				}
			}
			if (shouldSkip) {
				continue;
			}

			let isAllowed = true;
			try {
				const robotsTextPath = new URL(url).origin + "/robots.txt";
				const response = await fetch(robotsTextPath);

				const robots = robotsParser(robotsTextPath, await response.text());
				isAllowed = robots.isAllowed(url) ?? true; // respect robots.txt!
			} catch {}

			if (!isAllowed) {
				message.ack();
				continue;
			}

			try {
				const { numCloudflareLinks, screenshot } = await crawlPage(url);
				await env.CRAWLER_LINKS_KV.put(
					resultKey,
					numCloudflareLinks.toString(),
					{ metadata: { date: timestamp } },
				);
				await env.CRAWLER_SCREENSHOTS_KV.put(resultKey, screenshot, {
					metadata: { date: timestamp },
				});
				message.ack();
			} catch {
				message.retry();
			}
		}

		await browser.close();
	},
} satisfies ExportedHandler<Env, Message>;

8. 部署 Worker

要部署 Worker，请运行以下命令：

npx wrangler deploy

您已成功创建 Worker，可以将 URL 提交到队列进行爬取并将结果保存到 Workers KV。

要测试 Worker，您可以使用以下 cURL 请求截取本文档页面的截图。

Test with a cURL requestbash
curl <YOUR_WORKER_URL> \
  -H "Content-Type: application/json" \
  -d 'https://developers.cloudflare.com/queues/tutorials/web-crawler-with-browser-run/'

请参阅 完整教程的 GitHub 仓库 ↗，包括使用 Pages 部署的前端以提交 URL 并查看爬虫结果。

相关资源

这篇文档对您有帮助吗？