任务队列（SKIP LOCKED）
几乎每个业务系统迟早都需要一个「异步任务」能力：发邮件、生成报表、转码视频、调用第三方接口。这些任务不用立刻完成，但不能丢、不能重、还要能重试。于是一个经典的架构问题出现了——任务队列放在哪里？
答案是：很多场景下，PostgreSQL 自己就是最好的任务队列。你已经有了它，它有事务、有锁、有持久化、有可观测性；再引入一套独立的 RabbitMQ 或 Kafka，意味着多一个要部署、监控、备份、升级的系统组件，多一份数据一致性的对账负担（业务数据在库里，任务在 MQ 里，两边永远需要同步）。当任务量在每秒几百到几千以内、任务与业务数据强相关时，用数据库表 + 行级锁实现的队列，往往是最简单、最可靠的选择。
实现这个队列的核心是一条 SQL 技巧：FOR UPDATE SKIP LOCKED。它让多个 worker 在同一张队列表上「各抢各的任务，抢到谁就是谁」，互不阻塞、互不重复——这正是本页的主角。
前置知识
建议先阅读：软删除设计 与 SKIP LOCKED 行级锁。
基础概念
| 概念 | 英文 / 记法 | 说明 |
|---|---|---|
| 任务表 | job / task table | 用普通表存储待处理任务 |
| 生产者 | producer | 向队列表 INSERT 新任务的进程 |
| 消费者 / worker | consumer / worker | 轮询领取并处理任务的进程 |
| 领取 | claim / dequeue | 把任务从 pending 标记为 processing 的原子操作 |
FOR UPDATE SKIP LOCKED | — | 跳过已被其他事务锁定的行，避免 worker 互相等待 |
FOR UPDATE NOWAIT | — | 遇到锁立即报错而不是等待，用于快速失败的场景 |
| 可见时间 | visible_at | 任务下次可被领取的时间，实现延迟与重试退避 |
| 尝试次数 | attempts | 记录已尝试次数，控制最大重试 |
| 超时回收 | lease timeout | worker 崩溃后把超时的 processing 任务放回队列 |
| 幂等 | idempotency | 同一任务被执行多次结果不变，队列系统的必备属性 |
LISTEN/NOTIFY | — | 服务端推送通知，唤醒 worker 立即取任务 |
为什么行级锁能当队列用
队列的本质诉求有三条：
- 一条任务只被一个 worker 处理（互斥）；
- 多个 worker 并发领取时不互相阻塞（吞吐）；
- 领取是原子的（不会被中途打断、不会领到一半丢任务）。
普通的 SELECT ... FOR UPDATE 满足第 1、3 条，但不满足第 2 条：两个 worker 同时扫队列，都盯上了第一行，后到的事务必须等先到的提交（或回滚）才能继续——worker 越多排队越长，吞吐反而下降。
FOR UPDATE SKIP LOCKED 补上了第 2 条：扫到已被锁定的行时直接跳过，继续找下一行没被锁的。于是 N 个 worker 并发领取，各拿各的行，零等待、零重复。这个特性自 PostgreSQL 9.5 起可用，加上它诞生于数据库内核的事务与锁机制，天然继承了 ACID 的全部保障。
数据库队列 vs 消息中间件
| 维度 | PostgreSQL 队列表 | RabbitMQ | Kafka |
|---|---|---|---|
| 引入组件 | 无（已有） | 新增部署 | 新增部署 |
| 与业务数据的事务一致性 | 同一事务内原子 | 需要两段式/对账 | 需要两段式/对账 |
| 语义 | 精确一次领取，可重复消费需幂等 | at-least-once | at-least-once（分区有序） |
| 吞吐 | 每秒千级任务（可调） | 万级 | 十万级以上 |
| 重试/延迟 | 自由实现（SQL 即代码） | 内建死信/延迟插件 | 需要外建 |
| 可观测性 | SQL 直接查表 | 需要管理界面 | 需要监控体系 |
| 消息堆积能力 | 受磁盘与 VACUUM 影响 | 内存/磁盘队列 | 优秀（日志追加） |
| 运维复杂度 | 零增量 | 中 | 高 |
| 典型适用 | 业务内任务、与库数据强耦合 | 业务解耦、路由灵活 | 事件流、大吞吐管道 |
结论不是「数据库队列取代 MQ」，而是按耦合度与吞吐量选型：任务和业务数据同库同事务的（发邮件绑定「订单创建成功」），数据库队列一致性最好；纯粹的系统间事件分发、超大吞吐，才值得引入 MQ。
语法与用法
队列表设计
CREATE TABLE jobs (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
queue text NOT NULL DEFAULT 'default', -- 队列名（逻辑分流）
payload jsonb NOT NULL, -- 任务参数
status text NOT NULL DEFAULT 'pending', -- pending / processing / done / failed
attempts int NOT NULL DEFAULT 0, -- 已尝试次数
max_attempts int NOT NULL DEFAULT 5, -- 最大尝试次数
visible_at timestamptz NOT NULL DEFAULT now(), -- 何时可被领取（延迟/退避）
locked_by text, -- 领取者标识（主机+进程）
locked_at timestamptz, -- 领取时间（超时回收依据）
last_error text, -- 最近一次失败原因
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
-- 领取扫描的核心索引：只看「可领取」的任务
CREATE INDEX jobs_claim_idx
ON jobs (queue, visible_at)
WHERE status = 'pending';
-- 超时回收扫描的索引：只看「处理中」的任务
CREATE INDEX jobs_recover_idx
ON jobs (locked_at)
WHERE status = 'processing';
-- 已完成任务定期清理
CREATE INDEX jobs_done_idx ON jobs (updated_at) WHERE status IN ('done', 'failed');设计要点：
- 部分索引只覆盖「活跃」任务。绝大多数任务很快进入终态，
WHERE status = 'pending'的部分索引只有几百字节到几 MB，扫描永远飞快。 visible_at承担双重职责：入队延迟（visible_at = now() + interval '10 minutes'）与重试退避（visible_at = now() + backoff(attempts)）。status用 text 而不是 int，配合CHECK约束保证合法值，日志里人可直接读。locked_by建议放「主机名:进程号:随机串」，既定位进程又防混淆。
领取任务的核心 SQL
WITH picked AS (
SELECT id
FROM jobs
WHERE queue = 'emails'
AND status = 'pending'
AND visible_at <= now()
ORDER BY visible_at
LIMIT 10
FOR UPDATE SKIP LOCKED
)
UPDATE jobs j
SET status = 'processing',
attempts = j.attempts + 1,
locked_by = 'worker-1:31762',
locked_at = now(),
updated_at = now()
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.queue, j.payload, j.attempts;一条语句完成「挑选 + 锁定 + 标记 + 返回」，且整个过程在一个事务里原子完成。RETURNING * 把任务参数直接带回 worker，省一次回查。
FOR UPDATE SKIP LOCKED 写在 CTE 里：先挑出最多 10 个未锁定的候选 id（这一步锁定），再统一改成 processing。多个 worker 并发执行时，每个事务锁住自己挑中的行，其他事务 SKIP LOCKED 会绕开它们——这就是无阻塞并发领取的全部秘密。
完成与失败
-- 成功
UPDATE jobs
SET status = 'done', updated_at = now()
WHERE id = :job_id;
-- 失败：重试退避 or 终态
UPDATE jobs
SET status = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'pending' END,
visible_at = now() + power(2, attempts) * interval '30 seconds', -- 指数退避
last_error = :error_message,
locked_by = NULL,
locked_at = NULL,
updated_at = now()
WHERE id = :job_id AND status = 'processing';失败处理放在事务里与业务回滚协同：任务处理涉及的业务写入回滚了，任务本身标记失败等待重试；业务写入提交了，任务标记 done——业务数据与任务状态永远在同一事务里保持一致，这是数据库队列独有的优势。唯一要提防的是失败标记本身失效：连接抖动导致「标记 failed」的 UPDATE 没执行，任务会回到 ready 被重复领取——所以 worker 的收尾逻辑要对失败标记做幂等重试，领取条件的 WHERE 也按幂等设计。
延迟任务：visible_at 当定时器
很多队列需求里都有「10 分钟后再试」「明天早上 9 点执行」这类延迟语义。行模型里已经内置了答案：visible_at 就是任务的「最早可见时间」，入队时把它设到未来即可：
-- 延迟 10 分钟执行的任务
INSERT INTO jobs (queue, payload, visible_at)
VALUES ('emails', '{"to": "a@b.c"}', now() + interval '10 minutes');
-- 定时任务：明早 9 点（时间基准统一用 UTC 或固定时区落库，避免时区歧义）
INSERT INTO jobs (queue, payload, visible_at)
VALUES ('reports', '{"type": "daily"}',
date_trunc('day', now() + interval '1 day') + interval '9 hours');领取 SQL 的 WHERE visible_at <= now() 天然跳过未来任务，不需要额外的定时器表或独立的扫描逻辑。只有一点要注意：延迟任务的「准时」取决于 worker 的轮询间隔，而不是数据库——领取排序是 ORDER BY visible_at，worker 每 5 秒轮询一次，任务最多晚 5 秒被执行。需要更高精度时把轮询间隔调小，或配合下一节的 NOTIFY 做即时唤醒。最后提醒：visible_at 不需要触发器配合，重试与延期直接 SET visible_at = now() + interval '5 minutes'——把「下一次什么时候可见」的所有语义收敛到这一个列上。
顺带的架构红利：visible_at 让重试策略完全变成数据问题——失败第 N 次延迟多久（线性退避、指数退避、固定间隔）都只是 UPDATE 的取值函数，不需要任何额外的定时器基础设施。
LISTEN/NOTIFY 唤醒 worker
纯轮询的问题是有延迟（轮询间隔）或空转（没有任务时白查）。LISTEN/NOTIFY 让生产者在入队时顺带广播一个通知，worker 收到后立即取任务：
-- worker 会话：监听
LISTEN jobs_channel;
-- 生产者会话：入队 + 通知（同一事务）
BEGIN;
INSERT INTO jobs (queue, payload) VALUES ('emails', '{"to": "a@b.c"}');
NOTIFY jobs_channel, 'emails';
COMMIT;-- worker 端收到
Asynchronous notification "jobs_channel" with payload "emails" received from server process with PID 12345.NOTIFY 不携带大量数据（payload 上限约 8000 字节），正确用法是「通知有任务 + 队列名」，任务内容仍从表里取。worker 收到通知后执行领取 SQL；轮询作为兜底保留（比如每 5 秒一次），防止通知丢失。
详细说明
SKIP LOCKED 的工作原理
理解 SKIP LOCKED 的关键是回到 PostgreSQL 的 MVCC 与行锁模型：
- worker A 的事务执行
SELECT ... FOR UPDATE SKIP LOCKED，对扫到的行逐行加 row-level 锁。加锁成功的行进入结果集。 - worker B 几乎同时执行同样的查询。它扫到 worker A 已锁定的行时，不等待，直接跳过，继续扫后面的行，直到凑满
LIMIT n或扫完索引。 - A 提交（或回滚）后，它锁的行恢复「可锁」状态——但此时行已经被 A 改成
processing，不再满足status = 'pending'，不会再被任何人领走。 - 若 A 回滚，行仍是
pending，下一个 worker 可以领走——事务回滚自动把任务放回队列。
对比另外两种取法：
| 写法 | 并发行为 | 问题 |
|---|---|---|
| 裸 SELECT（无锁） | 都能看到同一批任务 | 重复消费 |
FOR UPDATE | 后到者阻塞等待 | worker 互相排队，吞吐塌陷 |
FOR UPDATE SKIP LOCKED | 后到者跳过 | 正解 |
FOR UPDATE NOWAIT | 后到者立即报错 | 不适合轮询领取，适合「必须立刻知道有没有锁」的治理类操作 |
SKIP LOCKED 的一个固有性质要注意：它不保证 FIFO。跳过被锁的行意味着后来者可能先取走后面的任务。绝大多数异步任务不要求严格先进先出；如果个别任务有顺序要求（如同一用户的操作必须串行），按 queue 或 payload->>'user_id' 拆分到不同队列，同队列内单 worker 串行。
幂等与事务边界
领取的原子性由 FOR UPDATE 保证，但「至少一次送达」（at-least-once）的语义仍然存在：worker 可能在处理到一半时崩溃，任务被超时回收后重新执行。因此处理逻辑必须幂等——同一任务执行两次，结果与执行一次相同。
幂等的通用做法是给任务或业务操作一个唯一键：
-- 每个任务带业务幂等键（如邮件任务用 message_id）
ALTER TABLE jobs ADD COLUMN idempotency_key text;
CREATE UNIQUE INDEX jobs_idem_uniq ON jobs (idempotency_key) WHERE status <> 'failed';
-- 业务表记录已处理键
CREATE TABLE processed_jobs (
idempotency_key text PRIMARY KEY,
processed_at timestamptz NOT NULL DEFAULT now()
);
-- 处理逻辑：先登记后干活，冲突即说明已处理过
BEGIN;
INSERT INTO processed_jobs (idempotency_key) VALUES (:key); -- 冲突则任务已处理
-- ... 业务写入 ...
UPDATE jobs SET status = 'done' WHERE id = :job_id;
COMMIT;事务边界的黄金法则：领取任务、业务写入、标记完成尽量在同一个数据库事务里。任务状态与业务数据的一致性由此免费获得；只有调用外部服务（发 HTTP 请求）时才不得不拆开事务，此时把「外部调用」设计成可重试 + 幂等（带唯一请求 ID），本地状态与外部效果用对账兜底。
重试与指数退避
重试的核心是「失败了别立刻再试」。指数退避让第 n 次重试的等待时间按倍数增长，给下游恢复的时间：
-- 延迟公式：base * 2^attempts（30s, 1m, 2m, 4m, 8m ...）
UPDATE jobs
SET visible_at = now() + power(2, LEAST(attempts, 10)) * interval '30 seconds'
WHERE id = :job_id;工程要点：
- 区分可重试与不可重试错误。参数非法、余额不足这类错误重试一万次也不会成功，直接进
failed终态；网络超时、第三方 5xx 才值得重试。 - 设置
max_attempts上限。超过上限进终态并告警，避免毒丸任务（poison job）无限重试耗尽 worker。 - 加抖动（jitter）。批量任务同时失败时，
base * 2^n会让它们同时再试，形成周期性尖峰；visible_at = now() + backoff * (1 + random())打散重试时间。 - 每次失败记录
last_error。重试十次失败的任务，排查时第一眼就能看到错误演进。
超时回收：worker 崩溃后任务怎么办
worker 领了任务（状态 processing）然后崩溃，任务就卡死了。解决思路是租约（lease）：locked_at 超过阈值的 processing 任务视为放弃，放回队列：
-- 回收超时任务（阈值按最长任务耗时 * 2 估算，例如 10 分钟）
UPDATE jobs
SET status = 'pending',
visible_at = now(),
locked_by = NULL,
locked_at = NULL,
last_error = 'lease expired, recovered'
WHERE status = 'processing'
AND locked_at < now() - interval '10 minutes'
RETURNING id, attempts;回收任务由任意 worker 周期执行（例如每分钟）。注意回收与 worker 仍在处理的慢任务之间的竞争：阈值必须大于最长任务耗时，否则「慢任务」会被回收导致重复执行——这也是幂等重要的另一个原因。回收时 attempts 已在领取时加过 1，重复领取会让重试计数自然增长，最终毒丸任务会被 max_attempts 拦下。
LISTEN/NOTIFY 的机制细节
NOTIFY 的三个重要性质：
- 与事务绑定。
NOTIFY只在包含它的事务提交后才送达；回滚则通知消失（与「任务行回滚」一致，不会出现假通知）。 - 只送到在线的监听者。worker 掉线期间的通知不会补发——所以必须有轮询兜底。
- 通知在连接断开后才能被新连接的 LISTEN 收到（同连接先发后听收不到历史），且同一事务内同一 channel + payload 的多次 NOTIFY 会合并为一次。
PHP 实现监听要用驱动层接口轮询 socket（PDO 没有 LISTEN 支持，通常用 pg_query('LISTEN ...') + pg_get_notify()），这比纯轮询复杂一档。务实的选择是：先用轮询起步，任务量大了再补 NOTIFY。
索引设计与性能
队列表的索引要精心设计，因为这张表的特点是高频率小事务读写 + 行快速失效：
- 领取查询走部分索引。
jobs_claim_idx只含 pending 行，无论表里有多少历史任务，扫描都只看活跃部分。 - 避免在
payload（jsonb）上做领取过滤。需要按 payload 过滤时加表达式索引（CREATE INDEX ON jobs ((payload->>'user_id')) WHERE status = 'pending'），而不是让领取查询全表扫 jsonb。 - 已完成任务定期删除。
done/failed任务保留一段时间（审计用）后批量删除，防止表无限膨胀。删除也要分批（每批几千行、单独事务），避免长事务与 WAL 峰值。 - 监控表膨胀。高频 UPDATE 的表容易膨胀，定期检查死元组比例，必要时调低该表的 autovacuum 阈值（
ALTER TABLE jobs SET (autovacuum_vacuum_scale_factor = 0.01)）。 - 吞吐参考。单实例上，领取 SQL 每次 1~2ms、每 worker 每秒可领数十个任务；8 个 worker 并发领取单表可达每秒数百到上千任务。瓶颈通常在任务本身的处理耗时，而不是领取 SQL。
与 RabbitMQ / Kafka 的选型再细化
把选型判断落成三个问题：
- 任务与业务数据是否同库？ 是 → 数据库队列（同一事务保证一致）。否 → 继续第 2 问。
- 峰值吞吐是否超过每秒数千任务？ 否 → 数据库队列足够，简单可靠优先。是 → 引入 MQ。
- 是否需要广播扇出 / 多消费者组重放？ 是 → Kafka 这类日志型系统是正解；否 → RabbitMQ 或数据库队列。
反过来的判断同样重要：不要为了「架构先进」引入 MQ。业务里 90% 的「发个通知、跑个报表」用队列表三小时就能上线，且与业务事务天然一致；MQ 的运维成本、对账成本、双写一致性问题会在未来数年里持续收费。
实战示例
示例一：并发领取演示（两个 psql 会话）
准备数据：
INSERT INTO jobs (queue, payload)
SELECT 'emails', jsonb_build_object('to', 'user' || g || '@x.com')
FROM generate_series(1, 6) AS g;会话 A：
BEGIN;
-- 领取（先提交领取结果前先看锁效果：不提交）
WITH picked AS (
SELECT id FROM jobs
WHERE queue='emails' AND status='pending' AND visible_at <= now()
ORDER BY visible_at LIMIT 2
FOR UPDATE SKIP LOCKED
)
UPDATE jobs j SET status='processing', locked_by='workerA', locked_at=now()
FROM picked WHERE j.id = picked.id
RETURNING j.id;
COMMIT; id
----
1
2
(2 rows)会话 B（在 A 未提交时执行同样的领取）：
id
----
3
4
(2 rows)B 领到的是 3、4——A 锁着的 1、2 被 SKIP LOCKED 跳过，B 没有等待、没有重复。这就是多 worker 并发领取的行为。
示例二：PHP worker 完整实现
<?php
// Queue/Worker.php —— 基于 PDO + PostgreSQL 的最小 worker
class Worker
{
public function __construct(
private PDO $pdo,
private string $name,
private string $queue = 'default',
private int $batch = 10,
private int $leaseSeconds = 600,
private int $pollIntervalUs = 1_000_000,
) {
$this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
}
/** 入队（生产者调用） */
public function enqueue(array $payload, string $idempotencyKey = null, int $delaySeconds = 0): int
{
$stmt = $this->pdo->prepare(<<<'SQL'
INSERT INTO jobs (queue, payload, idempotency_key, visible_at)
VALUES (:queue, :payload::jsonb, :key, now() + make_interval(secs => :delay))
RETURNING id
SQL);
$stmt->execute([
':queue' => $this->queue,
':payload' => json_encode($payload, JSON_UNESCAPED_UNICODE),
':key' => $idempotencyKey,
':delay' => $delaySeconds,
]);
return (int)$stmt->fetchColumn();
}
/** 主循环 */
public function run(callable $handler): never
{
while (true) {
$jobs = $this->claim();
if (!$jobs) {
usleep($this->pollIntervalUs);
continue;
}
foreach ($jobs as $job) {
$this->process($job, $handler);
}
}
}
/** 原子领取一批任务 */
private function claim(): array
{
$sql = <<<SQL
WITH picked AS (
SELECT id FROM jobs
WHERE queue = :queue
AND status = 'pending'
AND visible_at <= now()
ORDER BY visible_at, id
LIMIT :batch
FOR UPDATE SKIP LOCKED
)
UPDATE jobs j
SET status = 'processing',
attempts = j.attempts + 1,
locked_by = :name,
locked_at = now(),
updated_at = now()
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.payload, j.attempts, j.max_attempts
SQL;
$stmt = $this->pdo->prepare($sql);
$stmt->execute([
':queue' => $this->queue,
':batch' => $this->batch,
':name' => $this->name,
]);
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
/** 处理单个任务：成功标记 done，失败按策略重试或终态 */
private function process(array $job, callable $handler): void
{
try {
$handler(json_decode($job['payload'], true, 512, JSON_THROW_ON_ERROR));
$this->finish((int)$job['id'], 'done');
} catch (Throwable $e) {
$retryable = !($e instanceof DomainException); // 业务错误不重试
$exhausted = $job['attempts'] >= $job['max_attempts'];
$status = (!$retryable || $exhausted) ? 'failed' : 'pending';
// 指数退避 + 抖动：30s * 2^n，最大 1h
$backoff = min(30 * (2 ** (int)$job['attempts']), 3600)
* (1 + mt_rand(0, 100) / 100);
$stmt = $this->pdo->prepare(<<<'SQL'
UPDATE jobs
SET status = :status,
visible_at = now() + make_interval(secs => :backoff),
last_error = :err,
locked_by = NULL,
locked_at = NULL,
updated_at = now()
WHERE id = :id AND status = 'processing'
SQL);
$stmt->execute([
':status' => $status,
':backoff' => $backoff,
':err' => mb_substr($e->getMessage(), 0, 2000),
':id' => $job['id'],
]);
}
}
private function finish(int $id, string $status): void
{
$stmt = $this->pdo->prepare(
"UPDATE jobs SET status = :s, updated_at = now() WHERE id = :id AND status = 'processing'"
);
$stmt->execute([':s' => $status, ':id' => $id]);
}
/** 超时回收（由任一 worker 周期调用） */
public function recoverExpired(): array
{
$stmt = $this->pdo->prepare(<<<'SQL'
UPDATE jobs
SET status = 'pending', visible_at = now(),
locked_by = NULL, locked_at = NULL,
last_error = 'lease expired',
updated_at = now()
WHERE status = 'processing'
AND locked_at < now() - make_interval(secs => :lease)
RETURNING id
SQL);
$stmt->execute([':lease' => $this->leaseSeconds]);
return $stmt->fetchAll(PDO::FETCH_COLUMN);
}
}
// 启动：php worker.php
$pdo = new PDO('pgsql:host=127.0.0.1;dbname=app', 'app', 'secret', [PDO::ATTR_PERSISTENT => true]);
$worker = new Worker($pdo, name: gethostname() . ':' . getmypid(), queue: 'emails');
$worker->run(function (array $payload): void {
// 实际业务：发送邮件等。必须幂等。
mail($payload['to'], 'hello', 'world');
});示例三：可观测性查询
-- 队列健康总览
SELECT
queue,
count(*) FILTER (WHERE status = 'pending') AS pending,
count(*) FILTER (WHERE status = 'processing') AS processing,
count(*) FILTER (WHERE status = 'failed') AS failed,
max(visible_at) FILTER (WHERE status = 'pending') AS next_due,
avg(attempts) FILTER (WHERE status = 'done') AS avg_attempts_done
FROM jobs
GROUP BY queue; queue | pending | processing | failed | next_due | avg_attempts_done
--------+---------+------------+--------+-----------------------------+-------------------
emails | 42 | 3 | 1 | 2026-10-07 12:34:56.789+08 | 1.18
(1 row)-- 卡住的任务（领取超过租期仍未完成）
SELECT id, locked_by, locked_at, attempts, last_error
FROM jobs
WHERE status = 'processing'
AND locked_at < now() - interval '10 minutes';
-- 失败任务明细
SELECT id, queue, attempts, last_error, created_at
FROM jobs
WHERE status = 'failed'
ORDER BY updated_at DESC
LIMIT 20;监控阈值建议：pending 持续增长（积压）、processing 长时间不动（worker 挂）、failed 突增（下游故障）——三个信号分别接入告警。
这套查询走部分索引、开销极低，每 30 秒跑一次直接画进看板毫无压力——数据库队列的可观测性不需要额外组件，一张表的 SELECT 就够。
注意事项
SKIP LOCKED 必须与 FOR UPDATE 同用且在领取事务内。 领取 SQL 与后续处理必须在同一事务中完成状态翻转（或用一条 UPDATE 完成领取），裸 SELECT 出来的任务没有锁保护，等于没领。
领取事务不要混入耗时业务。 领取 → 标记 processing → 提交，然后才执行业务；否则长事务持有的行锁与连接占用会拖垮整个队列。
不保证严格 FIFO。 SKIP LOCKED 的跳过语义决定了顺序可能交错。有顺序要求的任务按键拆分队列、队列内串行处理。
任务必须幂等。 worker 崩溃、超时回收、重试都会导致重复执行。没有幂等的队列实现迟早双发邮件、重复扣款。用 idempotency_key + 已处理登记表是通用解法。
max_attempts 与不可重试错误都要有。 毒丸任务不限次重试会耗尽 worker；余额不足类业务错误重试无意义，直接终态 + 告警。
租期必须大于最长任务耗时。 否则慢任务被回收后重复执行；估算不出上限的任务，改为 worker 心跳续租（定期 UPDATE locked_at）。
队列表是高频 UPDATE 的表，关注膨胀。 调低该表的 autovacuum scale factor，已完成任务定期分批删除，防止索引与表无限增长拖慢领取扫描。
NOTIFY 不是可靠的投递保证。 掉线期间的通知不补发、重复通知会合并，轮询兜底不可省略。
jsonb payload 保持小而简单。 大对象存文件/对象存储，payload 里放引用；payload 里塞几 MB 的 jsonb 会让 TOAST 与 WAL 跟着遭殃。
连接要常驻并单独授权。 worker 使用持久连接与专用数据库角色，权限只限 jobs 表与业务所需表；连接池 transaction 模式下 LISTEN 不可用，worker 应用独立直连。
最佳实践
- 优先考虑数据库队列，超过能力边界再上 MQ。 判断标准：任务与业务同库同事务、峰值每秒数千以内、不需要广播/重放语义。
- 领取 SQL 固定为「CTE 挑选 + UPDATE 标记 + RETURNING」一条语句。 不自造多语句领取流程，原子性由单条 SQL 保证，团队代码风格也统一。
- 部分索引覆盖活跃状态。
WHERE status = 'pending'的索引让领取扫描与表大小解耦，这是队列表保持长期性能的关键。 - 幂等设计先于队列实现。 每个任务带业务幂等键，处理前登记已处理键，冲突即跳过；「至少一次」的投递语义与「精确一次」的业务效果分开对待。
- 重试用指数退避 + 抖动 + 上限。 退避基值与最大重试次数按下游承受力设定，终态任务必须有人工或自动的补偿路径。
- 超时回收作为后台例程常驻。 回收周期 1 分钟、租期按最长任务 2 倍估算，回收时写清
last_error，方便事后区分「真崩溃」与「慢任务」。 - 队列表单独调优 autovacuum 与删除策略。
autovacuum_vacuum_scale_factor = 0.01起步，done/failed 任务保留 7 天后分批删除，删除与 VACUUM 排进低峰。 - 监控三个信号：积压、停滞、失败率。
pending增长告警、processing超租期告警、failed比例超阈值告警；三个指标都从同一张表 SELECT，接入成本极低。 - 按业务域拆分队列名。
emails、reports、webhooks各自独立索引扫描与独立 worker，互不拖累；单一大队列是吞吐与故障隔离的双重风险。 - 外部调用拆出事务并做幂等。 数据库事务内只做本地写入；调用第三方时带唯一请求 ID，失败重试由队列机制承担，本地与远端的一致性用对账任务兜底。
- worker 无状态可水平扩展。 状态全在表里，worker 只是「领取-处理-标记」的循环；扩容即多起进程，缩容即杀进程，配合租约机制无需优雅关闭。
- 把「为什么不用 MQ」写进架构决策记录。 队列选型是最容易被反复挑战的决策，ADR 里写清吞吐、一致性、运维三个维度的量化对比，避免一年后无意义地重做一遍。
下一节
继续学习：多租户设计
参考链接
- PostgreSQL 官方文档：Row-Level Locks（FOR UPDATE / SKIP LOCKED）
- PostgreSQL 官方文档：NOTIFY（异步通知）
- PostgreSQL 官方文档：LISTEN
- PostgreSQL 官方文档：WITH Queries（CTE）
- PostgreSQL 官方文档：Partial Indexes（队列状态部分索引）
- PostgreSQL 官方文档：RETURNING（更新返回）
- 2ndQuadrant / EDB: What is SKIP LOCKED for in PostgreSQL 9.5?
- Cybertec: Implementing an efficient queue in PostgreSQL
- Stripe Blog: Idempotency keys（幂等设计经典文章）
- PostgreSQL 官方文档：Routine Vacuuming（高频更新表的维护）