SQLite 做工作流引擎：用单文件数据库实现持久化任务编排

当团队讨论「用什么做任务队列」时，答案通常是 Redis、RabbitMQ 或 Kafka。但 2026 年，越来越多的项目开始用一个出人意料的方案：SQLite。从 Litestream 到 Turso，从 rqlite 到 Obelisk——SQLite 正在从「嵌入式小数据库」进化为「生产级持久化引擎」。Hacker News 上一篇 “SQLite is all you need for durable workflows” 引发了 266 分的热议，背后是开发者对「过度工程化」的集体反思。本文将用完整代码演示如何用 SQLite 构建一个事务性工作流引擎，并与 Redis 方案做真实性能对比。

⚡ **关键结论：**SQLite 的 WAL 模式 + 事务保证，让它在中小规模场景下（< 10 万任务/天）完全可以替代 Redis + BullMQ，同时省去 Redis 的运维成本和数据丢失风险。

🏗️ 一、为什么 SQLite 能做工作流引擎？

1.1 传统方案的隐性成本

大多数团队在选型时只看到了 Redis 队列的「快」，却忽略了它的隐性成本：

💡 提示： 如果你的项目每天任务量 < 10 万，团队 < 20 人，SQLite 可能是性价比最高的方案。不要因为「大厂都用 Redis」就盲目选择——大厂的运维能力你不一定有。

1.2 SQLite 的三个核心优势

ACID 事务保证：SQLite 的事务是真正的原子操作——任务创建、状态更新、结果写入要么全部成功，要么全部回滚。Redis 的 MULTI/EXEC 无法做到这一点。

WAL（Write-Ahead Logging）模式：默认的 journal 模式在写入时会锁整个数据库，但 WAL 模式允许读写并发，写入性能提升 2-5 倍，读取几乎不受影响。

零运维：没有守护进程要监控，没有内存要调优，没有集群要配置。备份就是 cp data.db backup.db。恢复就是替换文件。

-- 开启 WAL 模式（只需执行一次，永久生效）
-- 这是 SQLite 用于生产环境的第一步，也是最关键的一步
PRAGMA journal_mode=WAL;

-- 设置 WAL 自动 checkpoint 阈值（默认 1000 页）
-- 建议生产环境设为 500-1000，避免 WAL 文件过大
PRAGMA wal_autocheckpoint=500;

-- 启用外键约束（SQLite 默认关闭）
PRAGMA foreign_keys=ON;

-- 设置忙等超时（毫秒），避免并发写入时立即返回 SQLITE_BUSY
PRAGMA busy_timeout=5000;

1.3 WAL 模式的工作原理

理解 WAL 模式是用好 SQLite 工作流引擎的关键。传统 journal 模式下，写入操作会：

1. 将原始页面复制到 rollback journal
2. 修改数据库文件
3. 删除 journal

这意味着写入期间任何读取都需要等待。而 WAL 模式完全相反：

1. 将修改写入 WAL 文件（追加写入，顺序 I/O）
2. 读取时从 WAL 文件 + 主数据库文件合并读取
3. Checkpoint 时将 WAL 中的修改合并到主数据库

传统模式:  读取 ──等待──▶ 写入完成 ──▶ 读取
WAL 模式:  读取 ─────────────────────▶ 读取（并发）
           写入 ─────────────────────▶ 写入（并发）

📌 记住： WAL 模式是 SQLite 生产化的前提条件。没有开启 WAL 的 SQLite 不适合做任何并发读写操作。

🔧 二、从零构建 SQLite 任务队列

2.1 数据库 Schema 设计

一个生产级的任务队列需要三张核心表：任务表、执行日志表、死信表。

-- 任务队列表：存储所有待执行和执行中的任务
CREATE TABLE IF NOT EXISTS jobs (
  id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
  queue TEXT NOT NULL DEFAULT 'default',     -- 队列名称，支持多队列
  type TEXT NOT NULL,                         -- 任务类型（如 'sendEmail', 'generateReport'）
  payload TEXT NOT NULL DEFAULT '{}',         -- JSON 格式的任务参数
  status TEXT NOT NULL DEFAULT 'pending',     -- pending | processing | completed | failed
  priority INTEGER NOT NULL DEFAULT 0,        -- 优先级，数值越大越先执行
  attempts INTEGER NOT NULL DEFAULT 0,        -- 已重试次数
  max_attempts INTEGER NOT NULL DEFAULT 3,    -- 最大重试次数
  run_at INTEGER NOT NULL DEFAULT (unixepoch('now')),  -- 计划执行时间（Unix 时间戳）
  started_at INTEGER,                         -- 开始执行时间
  completed_at INTEGER,                       -- 完成时间
  error TEXT,                                 -- 错误信息
  created_at INTEGER NOT NULL DEFAULT (unixepoch('now')),
  updated_at INTEGER NOT NULL DEFAULT (unixepoch('now'))
);

-- 复合索引：按队列 + 状态 + 优先级排序，这是 worker 拉取任务的核心查询
CREATE INDEX IF NOT EXISTS idx_jobs_fetch
ON jobs(queue, status, priority DESC, run_at ASC)
WHERE status = 'pending';

-- 索引：查询某个队列的任务统计
CREATE INDEX IF NOT EXISTS idx_jobs_queue_status
ON jobs(queue, status);

-- 执行日志表：记录每次执行的详细信息
CREATE TABLE IF NOT EXISTS job_logs (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  job_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
  attempt INTEGER NOT NULL,
  started_at INTEGER NOT NULL,
  completed_at INTEGER,
  duration_ms INTEGER,
  result TEXT,           -- 执行结果（JSON）
  error TEXT,
  created_at INTEGER NOT NULL DEFAULT (unixepoch('now'))
);

-- 死信表：超过最大重试次数的任务
CREATE TABLE IF NOT EXISTS dead_letters (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  job_id TEXT NOT NULL,
  queue TEXT NOT NULL,
  type TEXT NOT NULL,
  payload TEXT NOT NULL,
  error TEXT,
  attempts INTEGER NOT NULL,
  created_at INTEGER NOT NULL DEFAULT (unixepoch('now'))
);

2.2 任务队列核心实现（TypeScript）

以下是基于 better-sqlite3（同步 API，比 sqlite3 快 3-5 倍）的完整实现：

// job-queue.ts - SQLite 任务队列核心实现
import Database from 'better-sqlite3';
import { randomUUID } from 'crypto';

interface JobOptions {
  queue?: string;
  priority?: number;
  delay?: number;        // 延迟执行秒数
  maxAttempts?: number;
}

interface Job {
  id: string;
  queue: string;
  type: string;
  payload: string;
  status: string;
  priority: number;
  attempts: number;
  maxAttempts: number;
  runAt: number;
  startedAt: number | null;
  completedAt: number | null;
  error: string | null;
}

export class SQLiteJobQueue {
  private db: Database.Database;

  constructor(dbPath: string = 'jobs.db') {
    this.db = new Database(dbPath);
    this.db.pragma('journal_mode = WAL');
    this.db.pragma('busy_timeout = 5000');
    this.db.pragma('foreign_keys = ON');
    this.init();
  }

  private init(): void {
    this.db.exec(`
      CREATE TABLE IF NOT EXISTS jobs (
        id TEXT PRIMARY KEY,
        queue TEXT NOT NULL DEFAULT 'default',
        type TEXT NOT NULL,
        payload TEXT NOT NULL DEFAULT '{}',
        status TEXT NOT NULL DEFAULT 'pending',
        priority INTEGER NOT NULL DEFAULT 0,
        attempts INTEGER NOT NULL DEFAULT 0,
        max_attempts INTEGER NOT NULL DEFAULT 3,
        run_at INTEGER NOT NULL DEFAULT (unixepoch('now')),
        started_at INTEGER,
        completed_at INTEGER,
        error TEXT,
        created_at INTEGER NOT NULL DEFAULT (unixepoch('now')),
        updated_at INTEGER NOT NULL DEFAULT (unixepoch('now'))
      );

      CREATE INDEX IF NOT EXISTS idx_jobs_fetch
      ON jobs(queue, status, priority DESC, run_at ASC)
      WHERE status = 'pending';
    `);
  }

  // 添加任务到队列
  enqueue(type: string, payload: object, options: JobOptions = {}): string {
    const id = randomUUID();
    const {
      queue = 'default',
      priority = 0,
      delay = 0,
      maxAttempts = 3
    } = options;

    const runAt = Math.floor(Date.now() / 1000) + delay;

    this.db.prepare(`
      INSERT INTO jobs (id, queue, type, payload, priority, max_attempts, run_at)
      VALUES (?, ?, ?, ?, ?, ?, ?)
    `).run(id, queue, type, JSON.stringify(payload), priority, maxAttempts, runAt);

    return id;
  }

  // 原子性地获取并锁定一个任务（核心方法）
  dequeue(queue: string = 'default'): Job | null {
    // 使用事务确保原子性：查找 → 更新状态 → 返回
    const fetchJob = this.db.transaction(() => {
      const job = this.db.prepare(`
        SELECT * FROM jobs
        WHERE queue = ? AND status = 'pending' AND run_at <= unixepoch('now')
        ORDER BY priority DESC, run_at ASC
        LIMIT 1
      `).get(queue) as Job | undefined;

      if (!job) return null;

      // 立即标记为 processing，防止其他 worker 重复获取
      this.db.prepare(`
        UPDATE jobs
        SET status = 'processing', started_at = unixepoch('now'),
            attempts = attempts + 1, updated_at = unixepoch('now')
        WHERE id = ? AND status = 'pending'
      `).run(job.id);

      // 检查是否真的获取到了（可能被并发 worker 抢走）
      const updated = this.db.prepare(
        'SELECT * FROM jobs WHERE id = ? AND status = ?'
      ).get(job.id, 'processing') as Job | undefined;

      return updated || null;
    });

    return fetchJob();
  }

  // 标记任务完成
  complete(jobId: string, result?: object): void {
    this.db.prepare(`
      UPDATE jobs
      SET status = 'completed', completed_at = unixepoch('now'),
          updated_at = unixepoch('now')
      WHERE id = ?
    `).run(jobId);
  }

  // 标记任务失败，自动重试或移入死信队列
  fail(jobId: string, error: string): void {
    const job = this.db.prepare('SELECT * FROM jobs WHERE id = ?').get(jobId) as Job;

    if (!job) return;

    if (job.attempts >= job.maxAttempts) {
      // 超过最大重试次数，移入死信队列
      this.db.transaction(() => {
        this.db.prepare(`
          INSERT INTO dead_letters (job_id, queue, type, payload, error, attempts)
          VALUES (?, ?, ?, ?, ?, ?)
        `).run(job.id, job.queue, job.type, job.payload, error, job.attempts);

        this.db.prepare('DELETE FROM jobs WHERE id = ?').run(job.id);
      })();
    } else {
      // 指数退避重试：延迟 = 2^attempts 秒
      const retryDelay = Math.pow(2, job.attempts);
      this.db.prepare(`
        UPDATE jobs
        SET status = 'pending', error = ?,
            run_at = unixepoch('now') + ?,
            started_at = NULL,
            updated_at = unixepoch('now')
        WHERE id = ?
      `).run(error, retryDelay, jobId);
    }
  }

  // 获取队列统计信息
  stats(queue: string = 'default'): Record<string, number> {
    const rows = this.db.prepare(`
      SELECT status, COUNT(*) as count
      FROM jobs WHERE queue = ?
      GROUP BY status
    `).all(queue) as Array<{ status: string; count: number }>;

    const result: Record<string, number> = { pending: 0, processing: 0, completed: 0, failed: 0 };
    for (const row of rows) {
      result[row.status] = row.count;
    }
    return result;
  }

  // 清理已完成的任务（保留最近 N 条）
  cleanup(keepRecent: number = 1000): number {
    const result = this.db.prepare(`
      DELETE FROM jobs
      WHERE status = 'completed'
      AND id NOT IN (
        SELECT id FROM jobs
        WHERE status = 'completed'
        ORDER BY completed_at DESC
        LIMIT ?
      )
    `).run(keepRecent);
    return result.changes;
  }

  close(): void {
    this.db.close();
  }
}

2.3 Worker 执行器

有了队列，还需要一个 Worker 来消费任务：

// worker.ts - 任务执行器，支持并发控制和优雅退出
import { SQLiteJobQueue } from './job-queue';

type JobHandler = (payload: any) => Promise<any>;

export class Worker {
  private handlers = new Map<string, JobHandler>();
  private running = false;
  private concurrency: number;
  private activeJobs = 0;
  private pollInterval: number;

  constructor(
    private queue: SQLiteJobQueue,
    options: { concurrency?: number; pollInterval?: number } = {}
  ) {
    this.concurrency = options.concurrency || 1;
    this.pollInterval = options.pollInterval || 100; // 100ms
  }

  // 注册任务处理器
  register(type: string, handler: JobHandler): void {
    this.handlers.set(type, handler);
  }

  // 启动 Worker
  async start(): Promise<void> {
    this.running = true;
    console.log(`[Worker] Started with concurrency=${this.concurrency}`);

    while (this.running) {
      // 并发控制：最多同时执行 concurrency 个任务
      while (this.activeJobs < this.concurrency) {
        const job = this.queue.dequeue();
        if (!job) break;

        this.activeJobs++;
        this.processJob(job).finally(() => {
          this.activeJobs--;
        });
      }

      // 等待下一轮轮询
      await new Promise(r => setTimeout(r, this.pollInterval));
    }
  }

  private async processJob(job: any): Promise<void> {
    const handler = this.handlers.get(job.type);
    if (!handler) {
      console.error(`[Worker] No handler for job type: ${job.type}`);
      this.queue.fail(job.id, `No handler for type: ${job.type}`);
      return;
    }

    try {
      const payload = JSON.parse(job.payload);
      const result = await handler(payload);
      this.queue.complete(job.id, result);
      console.log(`[Worker] Job ${job.id} completed (${job.type})`);
    } catch (err: any) {
      this.queue.fail(job.id, err.message);
      console.error(`[Worker] Job ${job.id} failed: ${err.message}`);
    }
  }

  // 优雅退出：等待正在执行的任务完成
  async stop(): Promise<void> {
    console.log('[Worker] Stopping...');
    this.running = false;
    while (this.activeJobs > 0) {
      await new Promise(r => setTimeout(r, 50));
    }
    console.log('[Worker] Stopped');
  }
}

使用示例：

// app.ts - 完整使用示例
import { SQLiteJobQueue } from './job-queue';
import { Worker } from './worker';

const queue = new SQLiteJobQueue('my-app.db');

// 添加任务
queue.enqueue('sendEmail', { to: 'user@example.com', subject: 'Welcome' });
queue.enqueue('generateReport', { userId: 123, type: 'monthly' }, { priority: 10 });
queue.enqueue('sendReminder', { userId: 456 }, { delay: 3600 }); // 1 小时后执行

// 启动 Worker
const worker = new Worker(queue, { concurrency: 3 });

worker.register('sendEmail', async (payload) => {
  // 发送邮件逻辑
  console.log(`Sending email to ${payload.to}`);
  return { sent: true };
});

worker.register('generateReport', async (payload) => {
  // 生成报告逻辑
  console.log(`Generating ${payload.type} report for user ${payload.userId}`);
  return { reportUrl: '/reports/123.pdf' };
});

worker.start();

// 优雅退出
process.on('SIGINT', async () => {
  await worker.stop();
  queue.close();
  process.exit(0);
});

📊 三、性能基准测试与避坑指南

3.1 真实性能数据

我在一台 4 核 8GB 的 VPS 上做了基准测试，对比 SQLite 和 Redis + BullMQ：

⚡ 关键结论： 在单机 < 10 万任务/天的场景下，SQLite 的性能完全够用，且运维成本几乎为零。只有在需要多机分布式、或日任务量 > 100 万时，Redis 才有明显优势。

3.2 五个常见坑点

坑点 1：忘记开启 WAL 模式

默认的 DELETE journal 模式下，写入会锁整个数据库文件，并发读取会被阻塞。生产环境必须开启 WAL。

-- ❌ 错误：使用默认 journal 模式
-- 默认模式下写入会阻塞所有读取

-- ✅ 正确：开启 WAL 模式
PRAGMA journal_mode=WAL;

坑点 2：WAL 文件无限增长

如果没有设置 checkpoint，WAL 文件会持续增长。我在生产环境见过 WAL 文件膨胀到 10GB 的情况。

-- 设置自动 checkpoint，避免 WAL 文件过大
PRAGMA wal_autocheckpoint=1000;  -- 每 1000 页 checkpoint 一次

-- 手动 checkpoint（在低峰期执行）
PRAGMA wal_checkpoint(TRUNCATE);  -- checkpoint 后截断 WAL 文件

坑点 3：busy_timeout 设置太短

并发写入时，SQLite 会返回 SQLITE_BUSY 错误。默认的 busy_timeout 是 0（立即返回），这在高并发场景下会导致大量失败。

-- ❌ 错误：默认 busy_timeout=0，并发写入立即失败

-- ✅ 正确：设置 5 秒忙等
PRAGMA busy_timeout=5000;

坑点 4：没有正确处理事务边界

SQLite 的事务是数据库级别的，如果在事务中做了耗时操作（如网络请求），会阻塞其他所有写入。

// ❌ 错误：在事务中执行耗时操作
const result = db.transaction(() => {
  const job = dequeue();
  const response = await fetch('https://api.example.com'); // 阻塞！
  complete(job.id);
})();

// ✅ 正确：事务只做数据库操作，业务逻辑在事务外
const job = dequeue();          // 事务 1：获取任务
const response = await fetch('https://api.example.com');  // 业务逻辑
complete(job.id);               // 事务 2：更新状态

坑点 5：多人开发时的数据库文件锁定

SQLite 是文件级数据库，在多人开发时如果共享同一个 .db 文件（如放在共享目录），会遇到文件锁定问题。建议每个开发者使用独立的数据库文件，或使用 Docker 挂载。

3.3 生产环境最佳实践

✅ 备份策略：使用 Litestream 实时流式复制到 S3，RPO（恢复点目标）< 1 秒

✅ 监控指标：关注 WAL 文件大小、checkpoint 频率、SQLITE_BUSY 错误率

✅ 分库策略：不同队列使用不同的 .db 文件，避免单文件竞争

✅ 数据清理：定期清理已完成任务，保持数据库轻量

✅ Schema 迁移：使用 user_version pragma 管理版本，配合 better-sqlite3 的事务做原子迁移

// schema-migration.ts - 简单的数据库迁移方案
function migrate(db: Database.Database): void {
  const version = db.pragma('user_version', { simple: true });

  if (version < 1) {
    db.exec(`
      CREATE TABLE IF NOT EXISTS jobs (...);
      CREATE INDEX IF NOT EXISTS idx_jobs_fetch ON jobs(...);
    `);
    db.pragma('user_version = 1');
  }

  if (version < 2) {
    db.exec(`ALTER TABLE jobs ADD COLUMN tags TEXT DEFAULT '[]';`);
    db.pragma('user_version = 2');
  }
}

💡 总结与选型建议

SQLite 不是万能的，但在正确的场景下，它是最优解。

适合用 SQLite 做工作流引擎的场景：

• ✅ 单机部署的中小型应用（日任务量 < 10 万）
• ✅ 个人项目、Side Project、MVP 阶段
• ✅ 对数据持久性要求高、不能接受 Redis 内存丢失风险
• ✅ 运维资源有限的小团队

不适合的场景：

• ❌ 多机分布式部署（SQLite 是单写入者模型）
• ❌ 日任务量 > 100 万（WAL 写锁成为瓶颈）
• ❌ 需要消息确认、死信交换等高级消息队列功能
• ❌ 需要 Pub/Sub 实时通知模式

推荐的技术栈组合：

SQLite 的哲学是「简单胜于聪明」。在过度工程化泛滥的今天，选择一个零依赖、零运维、完全可控的方案，往往比追逐「大厂同款」更务实。正如 HN 网友的评论：“The best infrastructure is the one you don’t have to operate.”

相关工具推荐：

• better-sqlite3 — Node.js 最快的 SQLite 绑定
• Litestream — SQLite 实时流式备份
• Turso — SQLite 的分布式边缘数据库
• Drizzle ORM — TypeScript-first 的 SQLite ORM
• Obelisk — 基于 SQLite 的持久化工作流引擎