Crawlee 长任务断点续抓：中断后恢复任务并避免重复写入
结合 Crawlee 持久队列和 SQLite 幂等写入，演示写入后未确认时中断、重启继续、失败补采与结果对账，明确单进程恢复边界。
长任务中最危险的中断窗口，是数据已经写入、请求却还没被队列确认。重启后请求可能再次执行；如果输出只能追加，就会出现重复记录。反过来，先确认请求再写数据，又可能在中断时永久漏掉记录。
本文以两页图书目录为例，把持久请求队列与 SQLite 主键更新组合起来。正文包含一次明确的中断演示开关，用来说明重放为什么应该安全。示例面向单机、单进程任务，不宣称分布式 exactly-once。
准备输入地址
先在浏览器找到下列页面并确认内容，再从地址栏复制完整地址。服务或仓库地址从相应账户设置、官方项目的接入说明或克隆菜单取得。不要从示例输出中反推真实地址。
BOOKS_LIST_URL：Books to Scrape 总目录第一页地址。
在执行后续命令的同一个 Bash/WSL 终端中设置：
export BOOKS_LIST_URL='<paste-the-actual-books-list-url>'下文 JSON 或任务参数中的 <变量名> 是待替换标记，提交前填写刚复制的实际值；JSON 不会自动展开环境变量。结果示例中的 example.16yun.cn 只表示来源字段的形状，不是本文部署的演示站。
安装并把状态放进同一任务目录
使用 Node.js 24。它提供本文所需的 node:sqlite DatabaseSync；某些补丁版本可能显示模块实验性提示，具体能力以 Node SQLite 文档 为准。
mkdir resumable-catalog
cd resumable-catalog
npm init -y
npm install crawlee@3.18.2
export CRAWLEE_STORAGE_DIR="$PWD/resume-store"
export CRAWLEE_PURGE_ON_START=0同一任务恢复时保留同一目录、队列名和数据库。不要每次启动生成一个新目录，再把从头抓称作恢复。存储配置见 Crawlee 配置文档。
队列状态与商品状态分别保存
队列记录待处理或已处理的请求；数据库按 UPC 保存业务商品。请求 uniqueKey 控制调度身份，UPC 控制商品身份。它们不能互相替代。
本例把来源范围也写进数据库，防止无意中换入口后继续使用旧任务状态。只能有一个进程使用这份本地任务目录；并行部署需要重新设计队列与存储协调。
完整恢复示例
创建 resume.mts：
import { mkdirSync } from 'node:fs';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { CheerioCrawler, RequestQueue, NonRetryableError } from 'crawlee';
const storage = process.env.CRAWLEE_STORAGE_DIR;
if (!storage) throw new Error('请固定 CRAWLEE_STORAGE_DIR');
mkdirSync(storage, { recursive: true });
const db = new DatabaseSync(join(storage, 'catalog.sqlite'));
db.exec(`
CREATE TABLE IF NOT EXISTS products(
upc TEXT PRIMARY KEY,title TEXT NOT NULL,price_display TEXT NOT NULL,source_url TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS settings(k TEXT PRIMARY KEY,v TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS failures(
url TEXT PRIMARY KEY,label TEXT NOT NULL,page_no INTEGER,error_type TEXT NOT NULL
);
`);
const start = process.env.BOOKS_LIST_URL;
if (!start) throw new Error('Set BOOKS_LIST_URL before running');
const scope = JSON.stringify({ start, maxListPages: 2 });
const priorScope = db.prepare("SELECT v FROM settings WHERE k='scope'").get();
if (priorScope && priorScope.v !== scope) throw new Error('任务范围已变，请使用独立任务目录');
db.prepare("INSERT OR IGNORE INTO settings VALUES ('scope',?)").run(scope);
const queue = await RequestQueue.open('catalog-resume');
await queue.addRequest({ url: start, label: 'LIST', userData: { page: 1 } });
const repair = process.env.REPAIR_RUN;
if (repair) {
for (const row of db.prepare('SELECT * FROM failures').all()) {
await queue.addRequest({
url: String(row.url), label: String(row.label),
userData: { page: row.page_no },
uniqueKey: 'repair:' + repair + ':' + row.url,
});
}
}
const crawler = new CheerioCrawler({
requestQueue: queue,
maxConcurrency: 1,
maxRequestsPerCrawl: 100,
maxRequestRetries: 1,
async requestHandler({ request, $, enqueueLinks }) {
if (request.label === 'LIST') {
if (!$('article.product_pod h3 a').length) {
throw new NonRetryableError('列表没有商品');
}
await enqueueLinks({ selector: 'article.product_pod h3 a', label: 'DETAIL' });
if (Number(request.userData.page) < 2) {
if (!$('li.next a').length) throw new NonRetryableError('缺少下一页');
await enqueueLinks({
selector: 'li.next a', label: 'LIST', userData: { page: 2 },
});
}
db.prepare('DELETE FROM failures WHERE url=?').run(request.url);
return;
}
if (request.label !== 'DETAIL') throw new NonRetryableError('未知请求标签');
const fields = new Map<string, string>();
$('table.table-striped tr').each((_, row) => {
fields.set($(row).find('th').text().trim(), $(row).find('td').text().trim());
});
const upc = fields.get('UPC');
const title = $('.product_main h1').text().trim();
const price = fields.get('Price (incl. tax)');
if (!upc || !title || !price) throw new NonRetryableError('商品字段缺失');
const crash = process.argv.includes('--crash-once')
&& !db.prepare("SELECT v FROM settings WHERE k='crash_done'").get();
db.exec('BEGIN IMMEDIATE');
try {
db.prepare(`INSERT INTO products VALUES (?,?,?,?)
ON CONFLICT(upc) DO UPDATE SET title=excluded.title,
price_display=excluded.price_display,source_url=excluded.source_url`)
.run(upc, title, price, request.loadedUrl ?? request.url);
db.prepare('DELETE FROM failures WHERE url=?').run(request.url);
if (crash) db.prepare("INSERT INTO settings VALUES ('crash_done','1')").run();
db.exec('COMMIT');
} catch (error) {
db.exec('ROLLBACK');
throw error;
}
if (crash) {
console.log('模拟中断：数据已提交，处理函数尚未返回');
process.exit(99);
}
},
async failedRequestHandler({ request }, error) {
db.prepare(`INSERT INTO failures VALUES (?,?,?,?)
ON CONFLICT(url) DO UPDATE SET label=excluded.label,
page_no=excluded.page_no,error_type=excluded.error_type`)
.run(request.url, request.label ?? 'LIST',
Number(request.userData.page) || null, error.name);
},
});
try {
await crawler.run();
const products = db.prepare('SELECT count(*) AS count FROM products').get();
const failures = db.prepare('SELECT count(*) AS count FROM failures').get();
const finished = await queue.isFinished();
console.log({ products, failures, queue_finished: finished });
if (!finished || Number(failures?.count) > 0) process.exitCode = 1;
} finally {
db.close();
}正常运行可以直接执行 node resume.mts。若要观察本文讨论的中断窗口，在全新的独立任务目录按下面步骤进行：
node resume.mts --crash-once
node resume.mts第一条会在首次商品提交后以 99 退出；这是读者可自行执行的演示，不是本文的实测日志。第二条保留相同环境与目录，再继续处理。
为什么写入要先于请求确认
处理函数返回后，Crawlee 才按自己的流程确认请求。SQLite 先完成提交，因此中断后最坏情况是同一详情重放；主键更新把重复执行变成覆盖同一商品，而不是增加一行。
这不是把队列与数据库放进同一分布式事务。保证来自任务允许重放、业务写入幂等以及提交顺序。若处理还会发邮件、创建订单或扣费，必须分别解决那些副作用，不能只依靠数据库 UPC 主键。
中断后队列中的请求可能还受锁或租约影响，未必立即重新分配。检查待处理状态并按存储机制等待，不要为了看见进度新建队列，或者直接把所有请求标成已处理。队列接口见 RequestQueue 文档。
怎样确认恢复完成
练习目录若仍为每页二十本且无重复，业务结果应覆盖前两页的四十个 UPC。但真实检查还要看第二页样本、必填字段、失败表和 queue_finished，不能只凭总行数。
crash_done 标记和第一条商品在同一事务内保存。重启时这份演示状态不会要求再次中断；数据库里只有一条相同 UPC 的记录。不同任务要用不同目录，不能删掉商品表后保留旧的已处理队列。
程序再次运行而没有新请求，是队列已经完成的正常情况。需要更新商品内容时，应创建新的采集批次或明确补采，而不是期待已处理请求自动重新抓取。
失败补采与意外中断不是一回事
用尽重试后进入 failures 的请求，不能仅靠重启就假设会重新执行。修复网络或字段规则后，显式提供补采批次：
REPAIR_RUN=repair-1 node resume.mts代码为失败 URL 添加带补采批次的新 uniqueKey，保留原标签和列表页信息。同一 repair-1 重复启动不会无限添加同一补采请求；还需再次补查时使用新的明确批次。
成功后删除对应失败记录，失败则继续保留。SQLite 主键仍保护商品行不重复。不要只清空 failures 表让报告变绿，那会丢掉仍未处理的来源。
常见问题与修复
重启后像新任务。 检查工作目录、CRAWLEE_STORAGE_DIR、队列名与清理设置。先定位状态，不删除目录重试。
队列完成但数据库为空。 检查是否误删数据库或切换文件。队列与结果需要作为同一任务状态保留，不能各自独立恢复到不同时间。
数据库重复。 检查业务键是否稳定，以及实际存储是否仍只追加。请求 URL 去重无法代替业务写入幂等。
失败地址没有重跑。 使用显式补采批次，不把正常重启当成失败重置。
进程启动后暂时没有工作。 检查请求锁、存储配置和待处理状态。不要以一段安静日志推断所有商品已完成。
多个进程共享后行为异常。 本文约定单进程。先恢复独占目录，再设计适合多 Worker 的队列和事务方案，不在同一目录反复加锁碰运气。
可靠恢复允许重复执行，却不允许重复改变业务结果。把任务身份、提交顺序和补采规则都写清楚，比保存一个 processed_urls 文件更接近完整的长任务方案。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。