tl-pg-boss

tl-pg-boss

面向 Node.js、基于 PostgreSQL 的任务队列，支持恰好一次投递、Cron 调度和事务安全。

适用场景

• “为我的应用添加后台任务”
• “我需要一个任务队列，但已经在使用 Postgres”
• “设置 Cron 任务／计划任务”
• “处理支持重试的异步工作”
• 用户提到：任务队列、后台处理、任务调度、工作进程

考虑替代方案的场景

如果需要带有仪表板 UI 的托管后台任务，可以考虑 Trigger.dev。以下场景最适合使用 pg-boss：

• 你希望使用可自托管、原生支持 PostgreSQL 的任务队列
• 你已经在使用 Postgres，不想依赖外部服务
• 你需要事务性地将任务加入队列（与数据写入处于同一事务中）
• 你关注成本（pg-boss 免费，而托管服务按任务收费）

以下场景更适合使用 Trigger.dev：

• 你需要开箱即用、精致易用的仪表板
• 你希望使用支持自动扩缩容的托管基础设施
• 你的团队更喜欢托管式 SaaS 体验
• 你需要带可视化编排的复杂工作流

交付成果

• 补丁：安装并配置 pg-boss，以及带类型的任务处理器
• 产物：任务定义和适当的队列设置
• 决策：队列结构、任务模式和监控方案

---

要求

要求	版本
Node.js	22.12+
PostgreSQL	13+
权限	数据库上的 CREATE 权限

---

安装

pnpm add pg-boss

首次调用 start() 时会自动创建 pgboss 架构。无需手动迁移。

---

核心概念

SKIP LOCKED

PostgreSQL 的 SKIP LOCKED 无需分布式事务即可实现恰好一次投递。

始终调用 start()

重要：每个进程都必须调用 start()，包括生产者。

const boss = new PgBoss(connectionString);
await boss.start(); // Required in EVERY process

即使多个进程都调用 start()，也只有一个进程会运行监管程序。

每种任务类型对应一个队列

await boss.createQueue("send-email");
await boss.createQueue("process-image");

---

基本设置

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,
});

完整选项：参见 references/send-options.md

延迟任务

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

Cron 调度

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

取消调度（移除 Cron）

await boss.unschedule("daily-report");

调度管理：参见 references/schedule-management.md

---

工作进程模式

基本工作进程

await boss.work("my-queue", async ([job]) => {
  console.log(`Processing ${job.id}`);
  // Auto-completes on return, throw to fail
});

注意：即使 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: "[email protected]", subject: "Hi" });

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

TypeScript 模式：参见 references/typescript-patterns.md

---

队列配置

死信队列

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

-- Pending by queue
SELECT name, COUNT(*) FROM pgboss.job WHERE state = 'created' GROUP BY name;

-- Failed jobs
SELECT * FROM pgboss.job WHERE state = 'failed' ORDER BY completedon DESC LIMIT 20;

完整监控：参见 references/monitoring.md

---

易踩坑

易踩坑	解决方案
必须在所有地方调用 start()	生产者也需要调用
工作进程中的任务数组	使用 ([job])，而非 (job)
不支持 LISTEN/NOTIFY	仅支持轮询，请设置 pollingIntervalSeconds
架构需要 CREATE 权限	或使用 CLI：npx pg-boss migrate --dry-run
任务完成后无法再标记为失败	不要混用 work() 和手动调用 fail()
v10 中没有 pauseQueue()	使用 unschedule() 和直接 SQL
重启时会重新注册调度	代码会在初始化时调用 schedule()；取消调度仅为临时操作

---

最佳实践

1. 设置过期时间，防止任务成为僵尸任务：expireInMinutes: 30
2. 积极归档，并使用保留策略
3. 使用幂等处理器，并通过 upsert 写入
4. 优雅关闭：boss.stop({ graceful: true })

---

可观测性

参见可观测性，了解如何使用 prom-client 设置 Prometheus 仪表和计数器，以及针对队列积压和失败率的 Prometheus 告警规则。

---

测试模式

参见测试模式，了解如何对任务处理器进行单元测试、使用真实 PostgreSQL 数据库进行集成测试，以及在 vitest 中模拟 pg-boss。

---

Kubernetes 部署

有关包含资源限制、存活探针、preStop 钩子和 SIGTERM 优雅关闭处理程序的 worker Deployment 清单，请参阅 Kubernetes 部署。

---

验证

1. boss.start() 成功完成且无错误
2. boss.createQueue() 成功
3. boss.send() 返回作业 ID
4. Worker 处理作业
5. 作业状态变为 completed

---

参考资料

Quilted Skills

• triggerdotdev/skills/trigger-tasks — 工作流模式
• omer-metin/skills-for-antigravity/pg-boss — SKIP LOCKED 原则

第一方文档

• pg-boss GitHub — 官方仓库
• pg-boss API 文档 — API 参考
• pg-boss Dashboard — Web UI

PostgreSQL 资源

• PostgreSQL SKIP LOCKED — 锁定语义
• PostgreSQL Advisory Locks — 应用级锁
• 使用 PgBouncer 进行连接池管理 — 连接管理

替代方案

• Graphile Worker — PostgreSQL 队列替代方案
• Trigger.dev — 托管后台作业
• BullMQ — 基于 Redis 的替代方案

Skill 参考资料

• references/send-options.md — 完整的发送选项
• references/typescript-patterns.md — BaseJob、JobManager
• references/monitoring.md — Dashboard + SQL 查询
• references/advanced-patterns.md — Singleton、节流、发布/订阅
• references/schedule-management.md — Cron 计划、取消计划、暂停模式
• references/wordpress-migration.md — WP-Cron 和 Action Scheduler 映射

---

致谢

来源	作者	贡献
pg-boss	Tim Gilbert (@timgit)	核心库、官方文档
TypeScript Deep Dive	Shayan (@ImSh4yy)	BaseJob 类、JobManager 模式
pg-boss-admin-dashboard	Lyubomir Petrov (@lpetrov)	基于 JMESPath 的替代 Dashboard