用 Postgres 做任务队列：把 FOR UPDATE SKIP LOCKED 用对

运行后台任务不需要 Redis，也不需要消息代理。一张 Postgres 表加上 FOR UPDATE SKIP LOCKED，就能把每个任务一次只交给一个工作进程，在崩溃后依然可靠，并且让任务和它所属的数据处在同一个事务里。问题在于，最直观的第一版是错的：在我们的测试中，八个工作进程运行一条普通的「选出最早的任务，然后更新它」语句，同一行最多被领取了八次，10,000 个任务只完成了大约 2,100 个。本文复现这个 bug，构建修复它的队列（领取、完成、重试、回收），并介绍那些以后才会咬人的部分：工作期间一直打开的事务、LISTEN/NOTIFY、vacuum，以及轮询的工作进程会对空闲时挂起的计算实例造成什么影响。下文的每条语句和每段输出都来自本地的 PostgreSQL 17.11，也就是 Relyt One 运行的大版本。

一张表，每个任务一行。部分索引只覆盖工作进程要找的那些行，所以即使表里积累了很长的已完成任务历史，索引也依然很小。

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 'queued'
               CHECK (status IN ('queued', 'running', 'done', 'failed')),
  run_at       timestamptz NOT NULL DEFAULT now(),
  attempts     int         NOT NULL DEFAULT 0,
  max_attempts int         NOT NULL DEFAULT 5,
  locked_until timestamptz,
  last_error   text,
  created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX jobs_ready_idx ON jobs (queue, run_at) WHERE status = 'queued';

INSERT INTO jobs (payload)
SELECT jsonb_build_object('email_id', g) FROM generate_series(1, 10) AS g;

run_at 同时充当重试计划：当任务处于 queued 且 run_at <= now() 时，它就是就绪的。locked_until 是租约，稍后会讲到。

这是大多数人写出的第一个版本。因为它是一条语句，看起来是原子的：

UPDATE jobs
SET status = 'running', attempts = attempts + 1
WHERE id = (
  SELECT id FROM jobs
  WHERE status = 'queued'
  ORDER BY run_at
  LIMIT 1
)
RETURNING id, attempts;

在两个会话中同时运行它。会话 A 在一个事务中运行它，并且还在工作时，会话 B 运行了同一条语句：

-- session A (inside BEGIN, not committed yet)
 id | attempts
----+----------
  1 |        1

-- session B, returns after A commits
 id | attempts
----+----------
  1 |        2

两个会话都领取了任务 1，而任务 2 仍然是 queued。原因如下。子查询只运行一次，看到任务 1 是最早的 queued 任务，于是把值 id = 1 交给外层的 UPDATE。随后 B 必须等待 A 的行锁。在 READ COMMITTED 下，A 一提交，Postgres 就会针对该行的新版本重新评估 WHERE 子句。剩下的唯一条件是 id = 1，它仍然匹配，所以 B 更新了 A 已经拿走的任务。

实验室里的两个会话是一回事，所以我们用 pgbench 运行了八个工作进程，针对 10,000 个排队任务，每个工作进程领取 1,250 次：

pgbench -n -c 8 -j 8 -t 1250 -f naive.sql postgres

naive.sql 就是上面那条语句，只是改成 status = 'done'，所以每次领取同时也是一次完成。结束后：

 status | count | max_attempts
--------+-------+--------------
 done   |  2100 |            8
 queued |  7900 |            0

10,000 次领取只完成了大约 2,100 个任务（在两台机器上跑了八次，结果在 2,068 到 2,137 之间），有些任务被全部八个工作进程都领取了。在真实系统中，这就是同一封邮件发了八份。

锁住你选中的行，就能修复重复领取。把选取放进一个带 FOR UPDATE 的 CTE，再通过它来更新：

WITH next AS (
  SELECT id FROM jobs
  WHERE queue = 'default' AND status = 'queued' AND run_at <= now()
  ORDER BY run_at
  LIMIT 1
  FOR UPDATE
)
UPDATE jobs j
SET status = 'running', attempts = j.attempts + 1
FROM next
WHERE j.id = next.id
RETURNING j.id;

现在会话 B 会阻塞在任务 1 上。A 提交后，Postgres 重新检查任务 1，发现它已不再满足 status = 'queued'，于是丢弃它，转而取任务 2。这是正确的，但看看 B 的耗时：

 id
----
  2
(1 row)

UPDATE 1
Time: 1858.126 ms (00:01.858)

A 保持事务打开多久，B 就等了多久。工作进程一多，它们就会排在同一个首行后面，一次只能运行一个。

加两个词即可。SELECT 参考文档写道：“With SKIP LOCKED, any selected rows that cannot be immediately locked are skipped.”（使用 SKIP LOCKED 时，任何无法立即锁定的选中行都会被跳过。）文档也说明了它的用途：它给出的是不一致的数据视图，因此不适合通用场景，“but can be used to avoid lock contention with multiple consumers accessing a queue-like table.”（但可以用来避免多个消费者访问类队列表时的锁争用。）

在 A 仍持有任务 1 的情况下，B 立即拿到了任务 2：

 id
----
  2
(1 row)

UPDATE 1
Time: 2.512 ms

用 SKIP LOCKED 代替朴素语句，再跑一次同样的 pgbench：

 status | count | max_attempts
--------+-------+--------------
 done   | 10000 |            1

所有任务都完成了，每个任务恰好一次。

下面是此后我们要用的领取语句。它每次往返最多取 10 个任务（当每个任务都很快时，这一点很重要），并写入租约：

WITH next AS (
  SELECT id FROM jobs
  WHERE queue = 'default' AND status = 'queued' AND run_at <= now()
  ORDER BY run_at
  LIMIT 10
  FOR UPDATE SKIP LOCKED
)
UPDATE jobs j
SET status       = 'running',
    attempts     = j.attempts + 1,
    locked_until = now() + interval '5 minutes'
FROM next
WHERE j.id = next.id
RETURNING j.id, j.payload, j.attempts;

在一张有几千行的表上，EXPLAIN (COSTS OFF) 显示出你想要的形态：Limit 位于 LockRows 之上，LockRows 位于 jobs_ready_idx 上的索引扫描之上，因此工作进程只会触及就绪的行，并在取满 10 个后停止。

 Update on jobs j
   CTE next
     ->  Limit
           ->  LockRows
                 ->  Index Scan using jobs_ready_idx on jobs
                       Index Cond: ((queue = 'default'::text) AND (run_at <= now()))
                       Filter: (status = 'queued'::text)
   ->  Nested Loop
         ->  CTE Scan on next
         ->  Index Scan using jobs_pkey on jobs j
               Index Cond: (id = next.id)

这条领取语句有两种用法：

1. 边持锁边工作。 BEGIN，领取，执行工作，标记完成，COMMIT。崩溃时会回滚，任务重新变为空闲，这一点很有吸引力。
2. 租约。 领取后立即 COMMIT，状态为 status = 'running' 并带上 locked_until。在任何事务之外执行工作。再用第二个短事务把它标记为完成。

第一种适合耗时几毫秒的任务。凡是要调用 API、发送邮件或渲染文件的任务，都应使用租约。一个写入过数据（领取就写入过）然后停在那里等待的事务处于 idle in transaction 状态，只要它还打开着，vacuum 就无法移除在它开始之后变为死亡的行版本，而且这适用于该数据库中的任何表。一个保持打开的 REPEATABLE READ 事务即使只做过读取，也会造成同样的后果。

我们用这种方式让一次领取保持打开，让其他工作进程处理了 2,000 个任务，然后运行 VACUUM (VERBOSE) jobs：

tuples: 0 removed, 11992 remain, 2000 are dead but not yet removable

那个打开的事务提交后，再运行同一条命令：

tuples: 2001 removed, 8542 remain, 0 are dead but not yet removable

队列表上全是更新和删除，所以 vacuum 一旦受阻，它就是最先膨胀的表。打开的事务还会占用一条服务端连接；如果你位于事务模式的连接池后面，在你提交之前，这条连接都无法回到池中。短事务可以同时避免这两个问题。

有了租约，一个任务的整个生命周期就是四条短语句。工作进程在领取之后运行的那些语句会检查 status = 'running' 和 attempts（即领取时返回的值），因此工作进程只能修改仍归它所有的任务。

完成。 工作进程传回它从领取中拿到的 id 和尝试次数：

UPDATE jobs
SET status = 'done', locked_until = NULL
WHERE id = 1 AND status = 'running' AND attempts = 1;

失败并退避重试。 第一次失败后等 20 秒，第二次失败后等 40 秒，依次翻倍，上限为一小时；超过 max_attempts 后永久标记为 failed。在默认 5 次尝试的情况下，最长等待为 160 秒；只有当你调高 max_attempts 时，上限才会起作用。

UPDATE jobs
SET status       = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'queued' END,
    run_at       = now() + least(interval '1 hour', interval '10 seconds' * 2 ^ attempts),
    locked_until = NULL,
    last_error   = 'SMTP 451: try again later'
WHERE id = 2 AND status = 'running' AND attempts = 1
RETURNING id, status, attempts, run_at - now() AS retry_in;

 id | status | attempts | retry_in
----+--------+----------+----------
  2 | queued |        1 | 00:00:20

回收过期的租约。 如果工作进程在任务中途死掉，它的租约会到期，任何工作进程（或一个 cron 任务）都可以把任务放回队列。否则一个每次都会弄死工作进程的任务会无休止地回来，所以回收器同样遵守 max_attempts：

UPDATE jobs
SET status       = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'queued' END,
    locked_until = NULL,
    last_error   = coalesce(last_error, 'lease expired')
WHERE status = 'running' AND locked_until < now()
RETURNING id, status, attempts;

为什么尝试次数检查很重要。 假设处理任务 3 的工作进程并没有死，只是很慢。它的租约到期了，回收器把任务放回队列，另一个工作进程领取了它，于是 attempts 现在是 2。当慢的工作进程终于完成，并用 attempts = 1 运行完成语句时，什么也不会发生：

UPDATE 0

任务仍归它当前的所有者。如果没有这项检查，慢的工作进程会在第二个工作进程仍在运行该任务时把它标记为完成。你的工作仍然应该是幂等的，因为慢的工作进程产生的副作用已经发生了。这项检查只是让任务的状态保持真实。

选择一个比你最慢的正常任务更长的租约，或者让耗时长的任务在执行过程中自行延长租约；如果延长操作没有更新任何行，就停止，因为这个任务已经不归它了：

UPDATE jobs
SET locked_until = now() + interval '5 minutes'
WHERE id = $1 AND status = 'running' AND attempts = $2;

工作进程需要知道有活可干。最简单的办法是循环运行领取语句，没拿到任务就休眠。另一种办法是让 Postgres 通知它们：

CREATE FUNCTION jobs_notify() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  PERFORM pg_notify('jobs', NEW.queue);
  RETURN NULL;
END;
$$;

CREATE TRIGGER jobs_notify
AFTER INSERT ON jobs
FOR EACH ROW EXECUTE FUNCTION jobs_notify();

工作进程运行 LISTEN jobs，在收到通知时执行领取。NOTIFY 参考文档中的三个特性决定了你该如何使用它：

• 通知在提交时投递。 在事务内部，通知“are not delivered until and unless the transaction is committed.”（在事务提交之前不会投递，事务不提交就永远不会投递。）在我们的测试中，生产者在 13:08:30 插入了一个任务，并在 13:08:33 提交。监听者在 13:08:31 的检查什么也没看到，通知出现在它的下一次检查中。这正是你想要的，因为过早被唤醒的工作进程会找到一个它还看不到的任务。
• 通知不是队列。 只有在那一刻正在监听的会话才会收到通知，而且同一事务中相同的通知会被合并为一条。把通知当作「去看看」，永远不要把它当作任务本身，并保留一个频率较低的轮询作为兜底，以覆盖重连期间错过的通知。
• 载荷很小。 在默认配置下，载荷必须短于 8000 字节。发送队列名或一个 id，而不是任务本身。

还有一点：LISTEN 属于一条服务端连接，因此它需要直连连接，而不是经过事务模式连接池的连接。原因我们在 PgBouncer 事务池模式下会出哪些问题，以及如何逐一修复中讲过。生产者和领取查询可以使用经过连接池的连接。

在 Relyt One 上，计算实例在大约十分钟无事可做后会挂起，下一个连接到来时会将它恢复。对队列来说，关键在于什么算作「有事可做」。计算实例的活动监视器每半秒检查一次客户端会话：

• 任何不处于 idle 状态的会话（正在运行查询，或处于 idle in transaction）都算作当前的活动；
• 处于 idle 状态的会话，按其最后一次状态变化（也就是它最后一条查询结束的时刻）算作活动；
• 正在运行的 autovacuum 工作进程，或活跃的逻辑复制连接，也会让计算实例保持唤醒。

所以，一个每秒轮询一次的工作进程，或者每五分钟轮询一次的工作进程，都会让计算实例全天候保持唤醒，即使队列整晚都是空的。一个在工作期间保持事务打开的工作进程，会在整个任务期间让它保持唤醒。单独一个空闲连接则不会，因为它的最后一次状态变化离现在越来越远。由此可以得出三点。挂起会关闭所有连接，包括 LISTEN 连接，而一个立即重连的工作进程会把计算实例恢复。驱动或连接池发出的健康检查查询算作活动。频率高于每十分钟一次的兜底轮询会让计算实例一直开着。

这就给了你三个实实在在的选择：

1. 任务必须在几秒内开始。 接受工作进程运行期间计算实例始终开启。这是在任何平台上获得低延迟的代价。以你能接受的间隔轮询，或者使用 LISTEN 加兜底轮询。
2. 任务可以等一会儿。 不要全天候运行工作进程。按计划启动它们，让两次运行之间的间隔长于空闲超时；让它们用批量领取清空队列后退出。两次运行之间计算实例可以挂起，下一次运行的第一个连接会将它恢复。
3. 任务来自你自己的应用。 让负责把任务加入队列的那个请求同时触发一个工作进程（一个 Serverless 函数，或你所用平台任务运行器中的一个任务）。数据库仍然是状态、重试和租约的唯一可信来源，而且什么都不需要轮询。

分支在这里也能帮上忙。每个分支都有自己的计算实例，所以你可以让一组测试用的工作进程连到一个带有生产数据副本的分支上，猛烈地压测领取查询，而不去打扰生产环境的计算实例。

已完成的任务不需要留在热表中。按计划分批删除它们，或者把它们移到归档表：

DELETE FROM jobs
WHERE id IN (
  SELECT id FROM jobs
  WHERE status = 'done' AND created_at < now() - interval '7 days'
  LIMIT 10000
);

每一次领取、完成和重试都会写入一个新的行版本，所以这张表会持续产生死元组。关注 pg_stat_user_tables 中 jobs 的 n_dead_tup 和 last_autovacuum，并保持事务简短，让 vacuum 能真正移除它找到的死元组。

当任务与同一数据库中的行相关、吞吐量在每个任务一次 UPDATE 的量级、并且你希望入队与数据变更处于同一个事务中时，任务表就很合适。最后这一点是外部消息代理不借助额外机制就无法提供的。它不适合扇出给许多独立的消费者，不适合保留一条数据流供以后重放，也不适合更新带来的变动本身成为瓶颈的数据量。这些场景请使用消息代理或日志。

• 用 CTE 和 FOR UPDATE SKIP LOCKED 领取，绝不使用没有锁的 WHERE id = (SELECT ... LIMIT 1)
• 在就绪条件上使用部分索引
• 领取后立即提交；记录 locked_until；在事务之外执行工作
• 完成、失败和延长租约时使用 WHERE id = $1 AND status = 'running' AND attempts = $2，并检查受影响的行数
• 为过期租约运行回收器；用 max_attempts 限制重试次数
• 在直连连接上使用 LISTEN，把通知当作提示，保留兜底轮询
• 删除或归档已完成的任务；关注 n_dead_tup
• 了解什么会让 Serverless 计算实例保持唤醒，再决定工作进程如何等待

本文中的每条语句都是不依赖任何扩展的纯 PostgreSQL 17 SQL。创建一个免费的 Relyt One 项目，开一个分支，把两个 psql 会话连上去，看看 SKIP LOCKED 如何把不同的行分给它们。

免费开始使用