tl-pg-boss

用途

tl-pg-boss 是一个基于 PostgreSQL 的 Node.js 后台任务队列技能，使用 PostgreSQL 的 SKIP LOCKED 机制实现精确一次（exactly-once）投递，无需分布式事务。它适合在已经使用 Postgres 的应用中直接添加后台任务、任务调度、cron 作业或异步处理，避免引入额外的外部依赖。

主要能力

精确一次投递：借助 PostgreSQL 的 SKIP LOCKED 实现，无需分布式事务。
cron 定时调度：通过 boss.schedule() 注册 cron 表达式，boss.unschedule() 移除。
事务安全：支持与数据写入在同一事务中入队任务。
任务模式：支持普通任务、延迟任务（startAfter）、带重试与过期选项的任务、优先级、批量处理（batchSize）。
队列配置：支持死信队列（deadLetter）、保留策略（retentionMinutes）。
类型化任务：支持 TypeScript 泛型，如 boss.send<EmailJob>() 与 boss.work<EmailJob>()。
监控：提供 @pg-boss/dashboard 仪表盘，以及查询 pgboss.job 表的 SQL 示例。
集成示例：提供 Fastify 集成方式，包括 onReady 启动与 onClose 优雅关闭。

使用方式

安装：

pnpm add pg-boss

首次 start() 会自动创建 pgboss schema，无需手动迁移。

基础设置：

import { PgBoss } from "pg-boss";

const boss = new PgBoss({
  connectionString: process.env.DATABASE_URL,
  schema: "pgboss",
});

boss.on("error", console.error);
await boss.start();
await boss.createQueue("my-queue");

发送任务：

const jobId = await boss.send("my-queue", { userId: "123" });

带选项发送：

await boss.send("my-queue", payload, {
  retryLimit: 3,
  retryDelay: 60,
  expireInMinutes: 30,
  priority: 1,
});

延迟任务：

await boss.send("my-queue", payload, {
  startAfter: new Date(Date.now() + 60000),
});

cron 调度：

await boss.schedule("daily-report", "0 9 * * *", { type: "daily" });
await boss.unschedule("daily-report");

基础 Worker：

await boss.work("my-queue", async ([job]) => {
  console.log(`Processing ${job.id}`);
  // 返回时自动完成，抛出异常则失败
});

注意：即使 batchSize 为 1，回调也接收数组。

批量处理：

await boss.work("bulk-import", { batchSize: 10 }, async (jobs) => {
  for (const job of jobs) {
    await processItem(job.data);
  }
});

类型化任务：

interface EmailJob {
  to: string;
  subject: string;
}

await boss.send<EmailJob>("send-email", { to: "user@example.com", subject: "Hi" });

await boss.work<EmailJob>("send-email", async ([job]) => {
  await sendEmail(job.data.to, job.data.subject);
});

队列配置（死信队列与保留）：

await boss.createQueue("my-queue", { deadLetter: "my-queue-dlq" });
await boss.createQueue("my-queue-dlq");

await boss.createQueue("my-queue", { retentionMinutes: 60 * 24 });

Fastify 集成：

import Fastify from "fastify";
import { PgBoss } from "pg-boss";

const fastify = Fastify();
const boss = new PgBoss(process.env.DATABASE_URL);

fastify.decorate("boss", boss);

fastify.addHook("onReady", async () => {
  boss.on("error", fastify.log.error.bind(fastify.log));
  await boss.start();
  await boss.createQueue("my-queue");
  await boss.work("my-queue", handler);
});

fastify.addHook("onClose", async () => {
  await boss.stop({ graceful: true });
});

监控仪表盘：

pnpm add @pg-boss/dashboard
DATABASE_URL="postgres://..." npx pg-boss-dashboard

常用 SQL：

-- 按队列统计待处理任务
SELECT name, COUNT(*) FROM pgboss.job WHERE state = 'created' GROUP BY name;

-- 失败任务
SELECT * FROM pgboss.job WHERE state = 'failed' ORDER BY completedon DESC LIMIT 20;

适用场景

为已有 Postgres 的应用添加后台任务。
需要自托管、PostgreSQL 原生的任务队列，不想引入外部依赖。
需要事务性入队（与数据写入同一事务）。
需要 cron 定时任务或延迟任务。
成本敏感，希望使用免费方案而非按任务计费的托管服务。

限制与注意事项

要求 Node.js 22.12+、PostgreSQL 13+，数据库需具备 CREATE 权限。
每个进程都必须调用 start()，包括生产者进程；多个进程调用时只有一个运行监督。
每个任务类型应使用独立队列。
Worker 回调接收数组，应使用 ([job]) 而非 (job)。
不支持 LISTEN/NOTIFY，仅轮询，可设置 pollingIntervalSeconds。
任务一旦完成便无法再标记失败，不要混用 work() 与手动 fail()。
v10 没有 pauseQueue()，需使用 unschedule() 加直接 SQL。
调度在重启时会重新注册，unschedule 是临时的。
建议设置 expireInMinutes 防止僵尸任务，使用保留策略积极归档，处理器保持幂等，并使用 boss.stop({ graceful: true }) 优雅关闭。
若需要带仪表盘的托管后台任务，可考虑 Trigger.dev；若需要 Redis 方案，可考虑 BullMQ；PostgreSQL 队列替代方案可考虑 Graphile Worker。

验证步骤

boss.start() 无错误完成。
boss.createQueue() 成功。
boss.send() 返回任务 ID。
Worker 处理任务。
任务状态变为 completed。