批量任务跑到一半断了：用 SQLite 做去重、断点续跑和失败重试

30 秒速查

• 建 jobs 表：id、status、attempts、input_hash
• 入队用 INSERT OR IGNORE，同一条只进一次
• 领取用 BEGIN IMMEDIATE，先标 running
• 失败最多重试 3 次；卡住 15 分钟的自动回收

这篇教程解决什么问题

一个批量任务跑到第 200 条时断网、报错，或者你按了 Ctrl+C。第二天重跑，前面 199 条又处理了一遍：时间浪费了，API 费用也重复花了；更糟的是，有些步骤不幂等，重复执行会写出重复数据。

这篇用一个 SQLite 文件记录任务状态，解决四个问题：

• 去重：同一条业务数据只处理一次；
• 断点续跑：中断后从没完成的条目继续；
• 失败重试：失败的任务有限次重试，不拖垮整批；
• 可观察：随时能查出有多少条待处理、失败和完成。

它不依赖任何数据库服务，也不要求你把数据搬到一个新系统里。一个 .db 文件跟着脚本走。

为什么不用一个 JSON 状态文件

小批量任务用 JSON 记录进度是可以的，但一旦出现下面任一情况，就该换 SQLite：

• 进程可能被随时中断，JSON 写了一半会损坏；
• 两个 worker 同时更新，后写的会覆盖先写的；
• 任务上万条，每次启动都要把整个文件读进内存；
• 需要按状态筛选、统计、排序。

SQLite 的优势不是“性能更强”，而是每次修改都是一个事务：要么成功，要么完全没发生。这正好对应任务状态机需要的语义。

表结构

先建一张最小可用的表：

CREATE TABLE IF NOT EXISTS jobs (
  id TEXT PRIMARY KEY,
  kind TEXT NOT NULL,
  input_hash TEXT NOT NULL,
  payload TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  attempts INTEGER NOT NULL DEFAULT 0,
  result TEXT,
  error TEXT,
  updated_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs (status, updated_at);

字段含义：

字段	作用
id	业务主键，决定“什么算同一条任务”
kind	任务类型，方便一张表跑多种批处理
input_hash	输入内容的指纹，用来判断输入是否变化
payload	原始输入，失败重试时不用回头找源文件
status	pending / running / done / failed
attempts	已尝试次数，防止无限重试
result	成功结果，JSON 文本
error	最近一次失败原因
updated_at	用于领任务排序和回收卡死任务

id 怎么定很关键：

• 平台数据用“平台 + 原生 id”，例如 xhs:note:123456；
• 没有稳定 id 的内容，用规范化之后正文的哈希；
• 不要把“行号”当 id，源文件一改顺序就全乱了。

如果同一条内容允许修改后再处理，就把修改版本也放进 input_hash；如果不允许，内容哈希本身就是最好的去重键。

用 Node 内置的 SQLite

Node 22.5 起提供了 node:sqlite 模块，不需要安装第三方依赖。版本较旧时可能需要加 --experimental-sqlite 参数，先确认一下：

node --version
node -e "import('node:sqlite').then(() => console.log('ok'))"

建库和初始化：

import { DatabaseSync } from "node:sqlite";
import { createHash } from "node:crypto";

const db = new DatabaseSync("jobs.db");

db.exec(`
  PRAGMA journal_mode = WAL;
  PRAGMA busy_timeout = 5000;

  CREATE TABLE IF NOT EXISTS jobs (
    id TEXT PRIMARY KEY,
    kind TEXT NOT NULL,
    input_hash TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    attempts INTEGER NOT NULL DEFAULT 0,
    result TEXT,
    error TEXT,
    updated_at TEXT NOT NULL
  );

  CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs (status, updated_at);
`);

两个 PRAGMA 的作用：

• journal_mode = WAL 让读写互不阻塞，适合一边跑一边查询进度；
• busy_timeout = 5000 在锁冲突时最多等 5 秒，而不是立刻报错。

幂等入队

同一个任务可能被重复加入队列：脚本重跑、上游数据重复导出、手工补数据。入队操作必须幂等。

const MAX_ATTEMPTS = 3;
const STALE_MS = 15 * 60 * 1000;

function now() {
  return new Date().toISOString();
}

function hashInput(value) {
  return createHash("sha256")
    .update(JSON.stringify(value))
    .digest("hex")
    .slice(0, 16);
}

const upsertJob = db.prepare(`
  INSERT INTO jobs (id, kind, input_hash, payload, status, attempts, updated_at)
  VALUES (?, ?, ?, ?, 'pending', 0, ?)
  ON CONFLICT(id) DO UPDATE SET
    kind = excluded.kind,
    payload = excluded.payload,
    status = CASE
      WHEN jobs.input_hash <> excluded.input_hash THEN 'pending'
      ELSE jobs.status
    END,
    attempts = CASE
      WHEN jobs.input_hash <> excluded.input_hash THEN 0
      ELSE jobs.attempts
    END,
    input_hash = excluded.input_hash,
    updated_at = CASE
      WHEN jobs.input_hash <> excluded.input_hash THEN excluded.updated_at
      ELSE jobs.updated_at
    END
`);

function enqueue({ id, kind, payload }) {
  upsertJob.run(
    id,
    kind,
    hashInput(payload),
    JSON.stringify(payload),
    now(),
  );
}

这段 SQL 的行为：

• 任务不存在：插入一条 pending；
• 任务已存在且输入没变：什么都不改，已完成的任务不会重新跑；
• 任务已存在但输入变了：重置为 pending，重试次数清零。

这就是“改了输入才重跑，没改就不动”的完整逻辑。

领任务：先占住，再处理

单进程脚本可以直接“查一条、改状态”。多进程时，查询和更新之间可能被另一个进程插队，所以要用事务把两步包起来：

const selectJob = db.prepare(`
  SELECT * FROM jobs
  WHERE status IN ('pending', 'failed')
    AND attempts < ?
  ORDER BY updated_at
  LIMIT 1
`);

const markRunning = db.prepare(`
  UPDATE jobs
  SET status = 'running', attempts = attempts + 1, updated_at = ?
  WHERE id = ?
`);

function claimNext() {
  db.exec("BEGIN IMMEDIATE");

  try {
    const row = selectJob.get(MAX_ATTEMPTS);

    if (!row) {
      db.exec("COMMIT");
      return null;
    }

    markRunning.run(now(), row.id);
    db.exec("COMMIT");

    return { ...row, attempts: row.attempts + 1 };
  } catch (error) {
    db.exec("ROLLBACK");
    throw error;
  }
}

BEGIN IMMEDIATE 会立刻拿写锁，避免两个 worker 同时读到同一条 pending。先标记成 running 再处理，还有一个好处：进程如果中途被杀，这条任务会留下“运行中”的痕迹，后面可以回收。

完成与失败

const finishJob = db.prepare(`
  UPDATE jobs
  SET status = 'done', result = ?, error = NULL, updated_at = ?
  WHERE id = ?
`);

const failJob = db.prepare(`
  UPDATE jobs
  SET status = 'failed', error = ?, updated_at = ?
  WHERE id = ?
`);

主循环：

async function processJob(job) {
  const payload = JSON.parse(job.payload);

  // 换成你的真实业务逻辑
  return { id: job.id, length: payload.text.length };
}

async function main() {
  let done = 0;
  let failed = 0;

  for (;;) {
    const job = claimNext();
    if (!job) break;

    try {
      const result = await processJob(job);
      finishJob.run(JSON.stringify(result), now(), job.id);
      done += 1;
    } catch (error) {
      failJob.run(String(error.message ?? error).slice(0, 500), now(), job.id);
      failed += 1;
    }

    if ((done + failed) % 20 === 0) {
      console.log(`完成 ${done}，失败 ${failed}`);
    }
  }

  console.log({ done, failed });
}

main().catch((error) => {
  console.error(error);
  process.exit(1);
});

失败的任务不会阻塞后面的条目，只会被记下原因。再次运行脚本时，failed 且 attempts 小于 3 的任务会重新进入队列；超过 3 次的留在表里，等你排查。

回收卡死的任务

进程被强杀时，正在处理的那条会永远停在 running。启动时先做一次回收：

const resetStale = db.prepare(`
  UPDATE jobs
  SET status = 'failed', error = 'stale lease', updated_at = ?
  WHERE status = 'running' AND updated_at < ?
`);

function recoverStale() {
  const cutoff = new Date(Date.now() - STALE_MS).toISOString();
  return resetStale.run(now(), cutoff).changes;
}

在 main() 开头调用：

const recovered = recoverStale();
if (recovered > 0) console.log(`回收卡住的任务 ${recovered} 条`);

15 分钟的阈值适合单机批处理。如果你的任务单条就要跑半小时，把阈值调大；如果是单进程、重启后确定没有别的 worker 在跑，也可以在启动时把所有 running 直接改回 pending。多进程环境下不要这么做，会抢走别人正在处理的任务。

查进度和补数据

查看整体进度：

SELECT status, COUNT(*) AS n
FROM jobs
GROUP BY status
ORDER BY status;

看失败原因：

SELECT id, attempts, error, updated_at
FROM jobs
WHERE status = 'failed'
ORDER BY updated_at DESC
LIMIT 20;

手动重跑某一条：

UPDATE jobs
SET status = 'pending', attempts = 0, error = NULL, updated_at = '2026-10-24T00:00:00.000Z'
WHERE id = 'xhs:note:123456';

只补跑某个类型的任务：

UPDATE jobs
SET status = 'pending', attempts = 0
WHERE kind = 'summary' AND status = 'failed';

注意：手动改状态前先确认没有别的 worker 正在跑，否则可能和正在处理的任务撞车。

备份和清理

SQLite 的好处是备份简单。不要直接复制正在写入的 .db 文件，WAL 模式下还有 -wal 和 -shm 两个附属文件。用下面这条命令生成一致快照：

sqlite3 jobs.db "VACUUM INTO 'jobs-backup.db';"

没有安装命令行工具也可以在脚本里执行：

db.exec("VACUUM INTO 'jobs-backup.db'");

目标文件必须不存在，重复备份时换个带日期的文件名。

数据库变大后，可以清理已完成任务里的原始内容：

UPDATE jobs
SET payload = '{}'
WHERE status = 'done'
  AND updated_at < '2026-09-01T00:00:00.000Z';

保留状态和结果，删掉可能敏感的原文。清理后执行一次 VACUUM 回收空间。不要把二进制大文件塞进 SQLite，它擅长的是结构化状态，不是当文件仓库。

上线前检查

• id 的生成规则写进文档，确保不同脚本不会对同一条数据用不同主键；
• 所有入队都走同一个 enqueue，不要手工拼 SQL；
• 事务里不要做网络请求，领到任务后再处理；
• 失败原因要截断，不要把整段响应写进数据库；
• 给整批任务记一个批次 id，出问题能按批次回滚或重跑；
• 定期备份，至少保留最近两份快照。

做到这些，你的批量脚本就不再是“一次性的”，而是一个中断了能继续、重复跑也安全的工具。

下一步读什么

请求层用能重试的 AI API 客户端。

本地模型跑批时，用断网也能批量处理。