用 SKIP LOCKED 并发消费任务表避免队列锁等待的长事务

任务表充当轻量消息队列时，最常见的写法是 SELECT ... FOR UPDATE 配合 ORDER BY id LIMIT 1，再在同一个事务里把状态改成 processing。单进程下这套逻辑没有问题，一旦并发上去，问题就非常明显：多个 worker 的 SELECT 会排队等同一批行锁，谁先拿到锁谁先走，其余全部阻塞在 FOR UPDATE 上。更糟的是消费动作（调用外部 HTTP 接口、跑一次数据清洗）如果放在持锁事务里，事务可能持续几秒到几十秒，锁等待被成倍放大，表现为 LOCK等待 飙升、连接池打满、吞吐量不升反降。

PostgreSQL 9.5 引入的 FOR UPDATE SKIP LOCKED 正是为此而设计：遇到已被别的会话锁定的行，直接跳过，不再排队。配合 LIMIT 就能让 N 个 worker 各自拿到互不重叠的任务行，且不必等待。下面从锁行为、语句写法、事务边界、批量领取、索引与监控几个角度把它讲透，并给出可直接跑起来的 SQL 和常见报错处理。

先看清不加 SKIP LOCKED 时到底卡在哪

构造一张最小的任务表，然后开两个会话复现锁等待。

CREATE TABLE job_queue (
    id          bigserial PRIMARY KEY,
    kind        text        NOT NULL,
    payload     jsonb       NOT NULL DEFAULT '{}'::jsonb,
    status      text        NOT NULL DEFAULT 'pending',
    attempts    int         NOT NULL DEFAULT 0,
    run_after   timestamptz NOT NULL DEFAULT now(),
    locked_by   text,
    locked_at   timestamptz,
    created_at  timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT job_queue_status_chk
        CHECK (status IN ('pending', 'processing', 'done', 'failed'))
);

INSERT INTO job_queue (kind, payload)
SELECT 'send_mail', jsonb_build_object('uid', g)
FROM generate_series(1, 5) AS g;

会话 A 执行：

BEGIN;
SELECT id FROM job_queue
WHERE status = 'pending'
ORDER BY id
LIMIT 1
FOR UPDATE;
-- id = 1，事务不提交

会话 B 执行同样的语句，会一直挂住。此时在第三个会话里查 pg_stat_activity：

SELECT pid, wait_event_type, wait_event, state, left(query, 60) AS q
FROM pg_stat_activity
WHERE datname = current_database() AND state <> 'idle';

能看到会话 B 的 wait_event_type 为 Lock、wait_event 为 transactionid（等的是行锁持有者的事务 ID）。由于 B 的 SELECT ... FOR UPDATE 本身选择了第 1 行，它会安静地等下去，不会退而求其次去拿第 2 行。这就是队头阻塞：哪怕表里有 5 条任务、10 个 worker，也只有 1 个能推进。

注意：FOR UPDATE 拿的是行级排他锁，会阻塞其他 FOR UPDATE / FOR NO KEY UPDATE / FOR SHARE，并阻塞对该行的 UPDATE / DELETE。普通 SELECT 因 MVCC 不受影响，所以「查询不慢、领取全堵」是这种故障的典型特征。

换成 SKIP LOCKED：让 worker 各取各的

把领取语句改成三条关键子句组合：FOR UPDATE SKIP LOCKED + LIMIT，并且用 CTE 把「选中」和「更新」合成一条语句。

WITH picked AS (
    SELECT id
    FROM job_queue
    WHERE status = 'pending'
      AND run_after <= now()
    ORDER BY run_after, id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE job_queue j
SET status     = 'processing',
    attempts   = j.attempts + 1,
    locked_by  = 'worker-1',
    locked_at  = now()
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.kind, j.payload, j.attempts;

执行计划上，CTE 里的 SELECT 走 Index Scan 或 Bitmap Heap Scan，先在候选行上加锁，遇到锁冲突的行立即跳过；外层 UPDATE 只更新真正被选中的那一行。RETURNING 直接带回任务内容，省掉一次回表查询。

两个会话同时跑这条语句，结果一定是 id 不同。验证方法：在一个会话里 BEGIN 之后执行上面的 CTE 但不 COMMIT，第二个会话再执行，会拿到下一个 id 而不是阻塞。

-- 会话 A
BEGIN;
WITH picked AS (
    SELECT id FROM job_queue
    WHERE status = 'pending' ORDER BY id LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE job_queue j SET status = 'processing', locked_by = 'A'
FROM picked WHERE j.id = picked.id
RETURNING j.id;
-- 返回 id = 1，事务保持打开

-- 会话 B（立刻执行，不会等待）
BEGIN;
WITH picked AS (
    SELECT id FROM job_queue
    WHERE status = 'pending' ORDER BY id LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE job_queue j SET status = 'processing', locked_by = 'B'
FROM picked WHERE j.id = picked.id
RETURNING j.id;
-- 返回 id = 2

这里有一个必须讲清的语义细节：SKIP LOCKED 跳过的是「当前被锁的行」，不是「当前不可见的行」。如果 id=1 已经被 A 更新为 processing 且 A 还没提交，B 在 status = 'pending' 过滤下本身就看不到 id=1（MVCC 可见性），所以 B 会自然地选到 id=2。SKIP LOCKED 真正发挥作用的是另一类场景：A 只是锁了行但还没改状态（例如先锁再串行做别的事），或者 A 用的是 FOR NO KEY UPDATE 之外的锁模式。两种情况都会导致 B 在快照里仍能看到该行、但拿不到锁——SKIP LOCKED 让 B 不排队。

事务边界：消费逻辑绝不能留在持锁事务里

SKIP LOCKED 只解决了「领取时不排队」，并不解决「领取后事务过长」。如果 worker 在 UPDATE ... RETURNING 之后继续在同一事务里调用外部接口，任务行会长期处于 processing 且被行锁占用。此时其他 worker 靠 SKIP LOCKED 仍能继续领别的行，整体吞吐不受影响，但故障恢复会很麻烦：worker 进程崩溃时事务被回滚，任务回到 pending，这反而是好事；可如果客户端连接被中间层挂住（比如连接池不回收、TCP 半开），行锁会一直存在，只能靠 idle_in_transaction_session_timeout 兜底。

推荐做法是两段式：第一个短事务只做「领取 + 标记」，立即提交；消费在事务外执行；结束后用第二个短事务写结果。

-- 事务 1：领取，立即提交，持锁时间控制在毫秒级
BEGIN;
WITH picked AS (
    SELECT id FROM job_queue
    WHERE status = 'pending' AND run_after <= now()
    ORDER BY run_after, id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE job_queue j
SET status = 'processing', attempts = j.attempts + 1,
    locked_by = $1, locked_at = now()
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.kind, j.payload;
COMMIT;

-- 事务 2：worker 在事务外做完实际工作后写回
BEGIN;
UPDATE job_queue
SET status = 'done', locked_by = NULL, locked_at = NULL
WHERE id = $1 AND status = 'processing';
COMMIT;

把 WHERE id = $1 AND status = 'processing' 写进第二个事务，是为了防止「任务已被超时回收、又被另一个 worker 领走」时旧 worker 覆盖新状态。这类竞争在长任务场景下一定会发生，需要用状态条件做乐观保护。

顺便给出配套的会话级防护参数，防止异常连接把锁拖到天荒地老：

-- 对 worker 角色设置，单位毫秒
ALTER ROLE app_worker SET idle_in_transaction_session_timeout = '30s';
ALTER ROLE app_worker SET statement_timeout = '15s';
ALTER ROLE app_worker SET lock_timeout = '3s';

lock_timeout 配合 SKIP LOCKED 还有一个好处：万一某条语句忘了写 SKIP LOCKED（比如运维手工排查时），3 秒后就会报 ERROR: canceling statement due to lock timeout，而不是无限期挂着把连接池吃光。

批量领取：一次拿 N 条，减少往返

单条领取在高 QPS 下会让数据库承受大量短事务。把 LIMIT 1 改成 LIMIT N 即可，但要注意 ORDER BY ... LIMIT 与 SKIP LOCKED 的组合在大并发下可能出现「领到的行数少于 N」，这是正常现象：候选行被别的会话锁住后会被跳过，不会等待补齐。代码必须按「可能返回 0 到 N 行」处理，不要断言恰好 N 行。

WITH picked AS (
    SELECT id FROM job_queue
    WHERE status = 'pending' AND run_after <= now()
    ORDER BY run_after, id
    LIMIT 50
    FOR UPDATE SKIP LOCKED
)
UPDATE job_queue j
SET status = 'processing', attempts = j.attempts + 1,
    locked_by = $1, locked_at = now()
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.kind, j.payload, j.attempts;

大 LIMIT 会带来两个副作用：一是 UPDATE 一次性写多行，产生较大的 WAL 与死元组压力，autovacuum 跟不上时表膨胀明显；二是领取方在内存里堆一批任务，其中一部分可能在处理期间又超时。实践上单批 20~100 条比较稳，配合 worker 内部的小并发度（4~8）足够撑起每秒钟几百到上千任务的吞吐。

需要「一次只处理一种类型任务」时，在 WHERE 里加 kind = $2，并把索引建在 (kind, run_after, id) 上，避免全表扫。

索引与状态过滤：让 SKIP LOCKED 有可用的行可跳

SKIP LOCKED 的性能取决于「快速定位候选行」。任务表通常是状态单调推进的，pending 行占比很小，如果只在 status 上建普通 B-tree，随着 done 行累积，索引扫描会扫过大量已完成的键值，再回表过滤，效率迅速下降。用部分索引（partial index）只索引活跃状态：

CREATE INDEX idx_job_queue_pending
    ON job_queue (run_after, id)
    WHERE status = 'pending';

CREATE INDEX idx_job_queue_processing
    ON job_queue (locked_at)
    WHERE status = 'processing';

第一条索引直接服务领取语句，索引条目只包含待处理任务，随任务完成自动消失；第二条服务「超时回收」的扫描。建完之后用 EXPLAIN (ANALYZE, BUFFERS) 确认走的是 Index Scan using idx_job_queue_pending，并且 Rows Removed by Filter 接近 0。

EXPLAIN (ANALYZE, BUFFERS)
SELECT id FROM job_queue
WHERE status = 'pending' AND run_after <= now()
ORDER BY run_after, id
LIMIT 50
FOR UPDATE SKIP LOCKED;

如果执行计划出现 Seq Scan 且表很大，通常是统计信息过期或 run_after <= now() 的选择率估算失真，先 ANALYZE job_queue; 再看。

超时回收语句也一并给出，它是任务表队列的必备件——worker 崩溃后任务不能永远停在 processing：

UPDATE job_queue
SET status = CASE WHEN attempts >= 5 THEN 'failed' ELSE 'pending' END,
    locked_by = NULL, locked_at = NULL,
    run_after = now() + (power(2, attempts) * interval '1 second')
WHERE status = 'processing'
  AND locked_at < now() - interval '5 minutes'
  AND attempts < 5;

这里刻意避开了 FOR UPDATE SKIP LOCKED：回收操作是在「过期」范围内批量改状态，加锁等待的窗口很短（改的是 processing 行，通常没有并发写），而且回收器本身跑低频（每 30 秒一次），没必要为它引入跳过语义。真正要注意的是回收与正常 worker 的竞态——靠上一节提到的 AND status = 'processing' 条件更新来兜底。

常见报错与排障对照

把实际会碰到的错误信息和根因列成表，便于对照。

报错 / 现象	根因	处理
ERROR: FOR UPDATE cannot be applied to the nullable side of an outer join	领取语句用了 LEFT JOIN 且在外连接可空侧加锁	把 FOR UPDATE 改为 FOR UPDATE OF j 明确指定表别名，或去掉外连接
ERROR: FOR UPDATE is not allowed with DISTINCT clause	SELECT 里用了 DISTINCT / GROUP BY / 聚合函数	用 CTE 先取 id 加锁再关联原表，放弃在聚合结果上直接加锁
ERROR: canceling statement due to lock timeout	语句缺少 SKIP LOCKED，撞上长事务持有的行锁，超过 lock_timeout	补 SKIP LOCKED；同时查 pg_stat_activity 找持锁事务，必要时 terminate
ERROR: canceling statement due to statement timeout	领取语句扫描了大量失效候选行（通常卡在无索引的 Seq Scan）	补部分索引，ANALYZE 刷新统计，缩小 LIMIT
任务重复执行	worker 处理完但状态写回失败，超时回收把任务放回 pending	消费逻辑做幂等（业务唯一键 + ON CONFLICT DO NOTHING），状态写回加 AND status = 'processing'
任务丢失	领取事务已提交、worker 进程被杀，且没有超时回收器	必须有回收扫描，且 attempts 上限后转 failed 进入人工队列
ERROR: could not serialize access due to concurrent update	在 REPEATABLE READ 或 SERIALIZABLE 隔离级别下更新被并发修改的行	task queue 场景用 READ COMMITTED（默认值即可）；若必须高隔离级别，需在应用层重试

还有一类不易察觉的现象：加了 SKIP LOCKED 后 CPU 飙升但吞吐没涨。多数是「领取循环空转」——当表里没有可领任务时，worker 立刻返回 0 行然后又发一次查询，形成高频短查询。解决方式是拿到 0 行后 sleep（例如指数退避 100ms 起步，上限 2s），或者换成 LISTEN/NOTIFY 事件驱动加定时兜底。

可运行的并发验证脚本

用 pg_sleep 模拟慢消费，验证 SKIP LOCKED 下多 worker 不互相阻塞。先插入 20 条任务，然后开两个 psql 会话各执行下面这段（只改 locked_by 的值）：

DO $$
DECLARE
    r record;
    got int := 0;
BEGIN
    FOR i IN 1..10 LOOP
        WITH picked AS (
            SELECT id FROM job_queue
            WHERE status = 'pending' AND run_after <= now()
            ORDER BY run_after, id
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        ), upd AS (
            UPDATE job_queue j
            SET status = 'processing', attempts = j.attempts + 1,
                locked_by = 'worker-A', locked_at = now()
            FROM picked
            WHERE j.id = picked.id
            RETURNING j.id
        )
        SELECT id INTO r FROM upd;

        IF r.id IS NULL THEN
            RAISE NOTICE 'no pending job, exit at round %', i;
            EXIT;
        END IF;

        got := got + 1;
        -- 模拟消费耗时，之所以放在同一事务内，只是为了让锁驻留便于观察
        PERFORM pg_sleep(0.2);
    END LOOP;
    RAISE NOTICE 'worker-A picked % jobs', got;
END $$;

两个会话同时跑完，把 'worker-A' 换成 'worker-B' 再跑一次，统计能确认没有一条任务被两个 worker 同时领到：

SELECT locked_by, count(*) FROM job_queue
WHERE status = 'processing' GROUP BY locked_by;
-- 每个 locked_by 的行数之和应等于被领取总数，且不重叠

SELECT count(*) FROM job_queue WHERE attempts > 1;
-- 结果应为 0，说明没有重复领取

注意上面 DO 块把 pg_sleep 放在事务内是有意的——这样能直观看到行锁驻留、而另一会话仍能通过 SKIP LOCKED 继续推进。生产代码里请务必把它移出事务，按「短事务领取、事务外消费、短事务写回」的三段式来做。

小结

SKIP LOCKED 的价值不是「更快地加锁」，而是把排队换成跳过：worker 之间不再共享一个队头，锁等待从并发度的线性函数变成与候选行数有关的常数。配合部分索引限定候选集、短事务边界控制持锁时长、条件更新防御超时回收竞态，这张任务表就能在几十个 worker 下稳定跑起来。它仍不是通用 MQ：没有投递保证、没有分区消费组、没有背压，适合任务量在单表千万级以内、能接受至少一次投递语义的场景。任务规模继续增长时，再考虑按 kind 或哈希分表、把 done 行归档到历史表，保持活跃行规模可控。

FAQ

SKIP LOCKED 会漏掉任务吗？

不会。被跳过的只是「当前被其他会话锁住、无法立刻取得」的行，它在事务提交后重新变成可领取状态；只要它仍满足 status = 'pending' AND run_after <= now()，后续的领取请求一定能看到它。会「漏」的情况只有一种：没有任何 worker 再去查询它，那属于消费端停顿，不是 SKIP LOCKED 的语义问题。

用了 SKIP LOCKED 还需要 lock_timeout 吗？

建议保留。SKIP LOCKED 只对「行锁被占用」生效，遇到表级锁（如 ALTER TABLE、VACUUM FULL 拿的 ACCESS EXCLUSIVE）依旧会等。lock_timeout = '3s' 能让 worker 快速失败重试，而不是在维护窗口里堆积几百个等待连接。

为什么批量领取返回的行数比 LIMIT 少？

因为候选行在扫描过程中被并发会话锁住后被跳过，这是设计行为。把返回行数当作「至少 0、至多 N」处理；如果业务要求固定批量，改成领取循环直到凑够或候选耗尽。

REPEATABLE READ 下能用吗？

能执行，但会因快照固定而反复看到同一批已不可用的行，并且更新被并发修改的行时报 could not serialize access due to concurrent update，需要应用层重试。任务队列用默认的 READ COMMITTED 即可，这也是 PostgreSQL 的默认隔离级别。

任务表涨到上亿行怎么办？

先把 done / failed 行定期归档到 job_queue_history 并删除，保持活跃表在百万级；再按 kind 拆分表或按哈希分区（PARTITION BY HASH (kind)），让每个领取语句只扫自己那一片。分区后部分索引要建在分区上，父表只保留分区键条件。

能不能直接拿任务表当 MQ 用，不上 Kafka？

在任务量不大、且团队已经在用 PostgreSQL 做事务性写入的场景下完全可以：任务入队和业务数据变更可以放进同一个事务，这是外部 MQ 做不到的。代价是消费吞吐受单库写入能力限制，并且需要自己实现超时回收、重试计数、幂等——本文给出的 SQL 已经覆盖了前两项。超过每秒几千条时，建议换成专用消息系统，数据库只保留状态视图。