PHP小码农

Appearance

SKIP LOCKED 与 NOWAIT ​

前面几页讲的都是「等待」：等锁、等超时、等死锁检测。但有一类场景，等待本身就是错误的——多个消费者从同一张表抢任务时，没人愿意排队等别人做事。

PostgreSQL 为此在 SELECT ... FOR UPDATE 上提供了两个「不等待」的选项：NOWAIT（等不到就立刻报错）和 SKIP LOCKED（跳过被锁的行，取下一行）。其中 SKIP LOCKED 是把数据库当任务队列的关键能力，也是本页的核心内容。

前置知识

阅读本节前，建议先了解：行级锁（FOR UPDATE / FOR SHARE）、锁冲突矩阵

基础概念 ​

默认行为：排队等待 ​

SELECT ... FOR UPDATE 在遇到已被别人锁住的行时，默认行为是阻塞等待，直到对方提交或回滚，然后重新评估该行的可见性（EvalPlanQual），再决定是否返回。

sql
-- 会话 1：锁住任务 1
BEGIN;
SELECT * FROM tasks WHERE id = 1 FOR UPDATE;
-- 不提交

-- 会话 2：想锁同一行 → 会一直等下去
SELECT * FROM tasks WHERE id = 1 FOR UPDATE;
text
-- 会话 2 的状态
 wait_event_type | wait_event
-----------------+------------
 Lock            | transactionid

对队列场景来说，这种等待是完全多余的：会话 2 明明可以去处理任务 2，却在这里死等任务 1。

NOWAIT：等不到就报错 ​

NOWAIT 把「等待」换成「立即失败」：如果目标行已被锁定，立刻抛错，不进入等待队列。

sql
SET lock_timeout = 0;   -- 确保不是超时导致的失败

BEGIN;
SELECT * FROM tasks WHERE id = 1 FOR UPDATE NOWAIT;
text
ERROR:  could not obtain lock on row in relation "tasks"

SQLSTATE 是 55P03（lock_not_available）。注意它与 lock_timeout 触发的是同一个错误码，区别只在触发方式。

SKIP LOCKED：跳过被锁的行 ​

SKIP LOCKED 更进一步：它不报错、不等待，而是直接跳过那些已经被锁住的行，返回剩下的、可立即锁定的行。如果结果集为空（所有候选行都被锁），就返回 0 行。

sql
-- 会话 1：锁住任务 1
BEGIN;
SELECT * FROM tasks WHERE id = 1 FOR UPDATE;

-- 会话 2：SKIP LOCKED 会自动跳过任务 1，拿到任务 2
SELECT * FROM tasks
WHERE status = 'pending'
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED;
text
 id |  name   | status
----+---------+---------
  2 | task-2  | pending
(1 row)

这个「返回 2，而不是阻塞在 1」的行为，正是数据库即队列（Database as a Queue）模式的实现基础。

SKIP LOCKED 的语义不是「读一致性」

SKIP LOCKED 返回的结果是不一致的快照——它依赖「此刻谁锁着什么」这个瞬时状态。所以它只能用于「抢任务」这类语义上本来就允许跳过的场景，绝不能用于需要完整数据的逻辑（如对账、报表、统计）。

语法与用法 ​

完整语法位置 ​

NOWAIT / SKIP LOCKED 属于 SELECT 的锁定子句（Locking Clause），写在 FOR <mode> 之后：

text
[ WITH ... ]
SELECT [ ALL | DISTINCT ] select_list
FROM   table_expression
[ WHERE ... ]
[ GROUP BY ... ]
[ HAVING ... ]
[ WINDOW ... ]
[ ORDER BY ... ]
[ LIMIT ... ]
[ OFFSET ... ]
[ FOR { UPDATE | NO KEY UPDATE | SHARE | KEY SHARE }
      [ OF table_name [, ...] ]
      [ NOWAIT | SKIP LOCKED ] ]
sql
-- 四种组合
SELECT * FROM tasks WHERE status = 'pending' FOR UPDATE NOWAIT;
SELECT * FROM tasks WHERE status = 'pending' FOR UPDATE SKIP LOCKED;
SELECT * FROM tasks WHERE status = 'pending' FOR NO KEY UPDATE SKIP LOCKED;
SELECT * FROM tasks WHERE status = 'pending' FOR SHARE NOWAIT;
text
-- 语法错误示例（NOWAIT 与 SKIP LOCKED 不能同时用）
ERROR:  conflicting lock modes specified

与 LIMIT / ORDER BY 组合 ​

抢任务时最常见的写法是「按顺序取一条，跳过正在被处理的」：

sql
SELECT id, payload
FROM tasks
WHERE status = 'pending'
ORDER BY priority DESC, id
LIMIT 1
FOR UPDATE SKIP LOCKED;
text
 id |        payload
----+----------------------
 42 | {"job":"send_email"}
(1 row)

子句顺序不能乱

FOR UPDATE ... SKIP LOCKED 必须写在 LIMIT / OFFSET 之后。写成 FOR UPDATE SKIP LOCKED LIMIT 1 会直接语法报错。

详细说明 ​

SKIP LOCKED 的原子性 ​

SKIP LOCKED 最重要的性质是：「筛选 + 跳过已锁 + 加锁」这三件事在一条语句里原子完成。这正是它优于「先 SELECT 找空闲任务、再 UPDATE 抢占」的地方。

对比两种实现：

sql
-- 方案 A（错误）：先查后抢，存在竞争窗口
SELECT id FROM tasks WHERE status = 'pending' LIMIT 1;   -- 两个消费者可能拿到同一个 id
-- 应用层判断后
UPDATE tasks SET status = 'running' WHERE id = <id>;     -- 第二个消费者更新 0 行
sql
-- 方案 B（正确）：一条语句原子占领
UPDATE tasks
SET status = 'running', started_at = now(), worker = 'w1'
WHERE id = (
  SELECT id FROM tasks
  WHERE status = 'pending'
  ORDER BY id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
RETURNING id, payload;
text
 id |        payload
----+----------------------
  2 | {"job":"gen_report"}
(1 row)

方案 B 把「选中」和「锁定」合并成一个原子动作，两个并发消费者绝不会拿到同一行——因为第一个消费者锁住该行后，第二个消费者的子查询会直接跳过它。

用数据库当队列 ​

有了 SKIP LOCKED，一张普通表就能充当一个可靠的任务队列：

sql
CREATE TABLE task_queue (
  id          bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  queue       text NOT NULL DEFAULT 'default',
  payload     jsonb NOT NULL,
  status      text NOT NULL DEFAULT 'pending'
              CHECK (status IN ('pending', 'running', 'done', 'failed')),
  attempts    int  NOT NULL DEFAULT 0,
  max_attempts int NOT NULL DEFAULT 3,
  locked_by   text,
  locked_at   timestamptz,
  run_at      timestamptz NOT NULL DEFAULT now(),
  created_at  timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX idx_task_queue_pickup
  ON task_queue (queue, run_at, id)
  WHERE status = 'pending';
text
CREATE TABLE
CREATE INDEX

消费一条任务的完整语句：

sql
UPDATE task_queue
SET status     = 'running',
    locked_by  = 'worker-1',
    locked_at  = now(),
    attempts   = attempts + 1
WHERE id = (
  SELECT id
  FROM task_queue
  WHERE status = 'pending'
    AND run_at <= now()
    AND queue = 'default'
  ORDER BY run_at, id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
RETURNING *;
text
 id | queue   |         payload          | status  | attempts | locked_by | ...
----+---------+--------------------------+---------+----------+-----------+----
  1 | default | {"job":"send_email"}     | running |        1 | worker-1  | ...
(1 row)

为什么 UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED) 是标准写法

直接在 UPDATE 上写 LIMIT 是语法错误（UPDATE 不支持 LIMIT）。所以标准做法是用子查询 + LIMIT 1 + FOR UPDATE SKIP LOCKED 选出唯一目标行，再让外层 UPDATE 更新它。子查询里拿到的行锁会一直持有到事务提交，保证从「选中」到「改状态」之间不会被别人抢走。

生产者-消费者完整实现 ​

下面是可运行的完整队列实现，包括入队、消费、成功、失败重试与超时回收。

sql
-- 1) 入队
CREATE OR REPLACE FUNCTION enqueue(p_queue text, p_payload jsonb)
RETURNS bigint
LANGUAGE sql AS $$
  INSERT INTO task_queue (queue, payload)
  VALUES (p_queue, p_payload)
  RETURNING id;
$$;

-- 2) 消费一条（原子抢占 + 返回任务）
CREATE OR REPLACE FUNCTION dequeue(p_queue text, p_worker text)
RETURNS TABLE (id bigint, payload jsonb, attempts int)
LANGUAGE sql AS $$
  UPDATE task_queue
  SET status    = 'running',
      locked_by = p_worker,
      locked_at = now(),
      attempts  = task_queue.attempts + 1
  WHERE task_queue.id = (
    SELECT q.id
    FROM task_queue q
    WHERE q.status = 'pending'
      AND q.queue  = p_queue
      AND q.run_at <= now()
      AND q.attempts < q.max_attempts
    ORDER BY q.run_at, q.id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
  )
  RETURNING task_queue.id, task_queue.payload, task_queue.attempts;
$$;

-- 3) 成功标记完成；4) 失败则按剩余次数重排或标记失败；5) 回收僵死任务
UPDATE task_queue SET status = 'done', locked_by = NULL, locked_at = NULL WHERE id = 1;

UPDATE task_queue
SET status   = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'pending' END,
    run_at   = now() + interval '30 seconds',
    locked_by = NULL,
    locked_at = NULL
WHERE id = 1;

UPDATE task_queue
SET status = 'pending', locked_by = NULL, locked_at = NULL
WHERE status = 'running'
  AND locked_at < now() - interval '10 minutes';
text
CREATE FUNCTION
UPDATE 1
UPDATE 1
UPDATE 0

必须把「抢占」和「改状态」放在同一个事务里

dequeue 里的 UPDATE 本身就在一个事务里（隐式或显式），行锁一直持有到事务结束。不要先调用 dequeue 提交，再在业务层慢慢处理——那样一旦 worker 崩溃，任务会永远卡在 running。正确做法是：抢占 → 在一个事务内完成处理 → 提交；或者抢占后立即提交并在业务完成后更新状态，同时用「僵死回收」兜底。

NOWAIT 的适用场景 ​

NOWAIT 与 SKIP LOCKED 的取舍可以用一句话概括：

选项遇到已锁行时适合场景
默认（都不写）等待对方结束必须处理这一行、且能接受等待
NOWAIT立刻报错 55P03宁可失败也不要排队（如用户交互、短超时 API）
SKIP LOCKED跳过，取下一行队列、分片处理、多个 worker 分摊任务
sql
-- 场景：用户点击「立即提现」，不想让请求卡在锁等待上
BEGIN;
SET LOCAL lock_timeout = '200ms';
SELECT * FROM accounts WHERE id = 100 FOR UPDATE NOWAIT;
UPDATE accounts SET balance = balance - 100 WHERE id = 100;
COMMIT;
text
ERROR:  could not obtain lock on row in relation "accounts"

应用层捕获 55P03，提示用户「账户正忙，请稍后重试」，而不是让请求挂住。

分片与批处理 ​

SKIP LOCKED 不只能用于「逐条」。处理大批数据时，可以让多个 worker 各领一批互不重叠的行：

sql
-- 每个 worker 领取 100 行做处理
WITH claimed AS (
  SELECT id
  FROM task_queue
  WHERE status = 'pending'
  ORDER BY id
  LIMIT 100
  FOR UPDATE SKIP LOCKED
)
UPDATE task_queue t
SET status = 'running', locked_by = 'batch-worker-1', locked_at = now()
FROM claimed c
WHERE t.id = c.id
RETURNING t.id;
text
  id
------
    1
    2
  ...
(100 rows)

-- 多个 worker 同时运行时，各自拿到的是「别人没锁住的 100 行」，集合天然不重叠：
-- worker-1 领取: 1..100      worker-2 领取: 101..200      worker-3 领取: 201..300

批处理要配批内短事务

LIMIT 100 会让你在一个事务里锁住 100 行。如果处理很慢，这 100 行会长时间不可用。更好的做法是每条任务一个事务（LIMIT 1），或者至少把批大小控制在几百行以内，避免「大批量锁」演变成新的锁热点。

报表与只读场景为什么不能用 ​

再次强调 SKIP LOCKED 的语义陷阱。假设用它做对账：

sql
-- 错误！用 SKIP LOCKED 做统计
SELECT count(*) FROM orders FOR UPDATE SKIP LOCKED;
text
 count
-------
   48213     -- 实际可能是 50000 行，被锁住的那部分被静默跳过了

这个结果既不精确、也不可重复：换个时刻执行会得到不同的数字。同理，以下场景都不允许使用 SKIP LOCKED：

场景为什么不能用
统计 / 汇总 / 对账结果会遗漏被锁的行
数据迁移 / 全量同步漏数据且难以察觉
分页展示每页内容不确定，用户会看到重复或缺失
唯一性校验可能漏掉并发的冲突行

正确做法是：统计就用普通 SELECT（只加 ACCESS SHARE 表锁，不需要行锁），且不加任何锁定子句。

与 READ COMMITTED 的配合 ​

SKIP LOCKED 在高并发下能被广泛使用，前提是 PostgreSQL 的 READ COMMITTED（读已提交） 语义：事务只能看到自己开始时（严格说是每条语句开始时）已提交的数据。

sql
-- worker-1 与 worker-2 都在 READ COMMITTED 下
-- worker-1: 抢占任务 1，未提交
BEGIN;
SELECT * FROM tasks WHERE status = 'pending' ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;
-- 返回 id=1

-- worker-2: 同一时刻
SELECT * FROM tasks WHERE status = 'pending' ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;
-- 跳过 id=1（被锁），返回 id=2
text
-- worker-2 结果
 id | status
----+---------
  2 | pending
(1 row)

两个 worker 分别拿到 1 和 2，没有重复、没有阻塞——这就是「数据库当队列」能成立的核心机制。

与 SERIALIZABLE 的限制 ​

在 SERIALIZABLE 隔离级别下，PostgreSQL 使用 SSI（Serializable Snapshot Isolation，可串行化快照隔离），会读写谓词锁（Predicate Lock）。这时 SKIP LOCKED 的行为有重要限制：

  • SKIP LOCKED 仍然可用，但不会阻塞在谓词锁上——它只会跳过「行锁冲突」的行。
  • 因此在高并发下，SERIALIZABLE + SKIP LOCKED 更容易出现序列化失败（SQLSTATE 40001），应用必须重试。
  • 队列这类「本来就不要求串行化语义」的场景通常不需要 SERIALIZABLE；用 READ COMMITTED 反而更合适。
sql
BEGIN ISOLATION LEVEL SERIALIZABLE;
SELECT * FROM tasks
WHERE status = 'pending'
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED;
COMMIT;
text
ERROR:  could not serialize access due to read/write dependencies among transactions
DETAIL:  Reason code: Canceled on identification as a pivot, during write.
HINT:  The transaction might succeed if retried.

队列不需要 SERIALIZABLE

SKIP LOCKED 的语义本身就是「接受不确定性」——跳过哪些行取决于瞬时锁状态。这与 SERIALIZABLE 追求的「全序一致」目标相反。生产环境中的任务队列应当使用 READ COMMITTED（默认级别）。

锁范围与索引的关系 ​

FOR UPDATE SKIP LOCKED 的性能高度依赖索引。如果查询条件没有可用索引，PostgreSQL 必须扫描所有候选行，对每一行尝试加锁并检查是否被跳过——行数一多，SKIP LOCKED 就会变得非常慢。

sql
-- 好：有部分索引，能直接定位到 pending 的行
CREATE INDEX idx_task_queue_pickup
  ON task_queue (queue, run_at, id)
  WHERE status = 'pending';

EXPLAIN (ANALYZE, BUFFERS)
SELECT id FROM task_queue
WHERE status = 'pending' AND queue = 'default'
ORDER BY run_at, id
LIMIT 1
FOR UPDATE SKIP LOCKED;
text
                                          QUERY PLAN
---------------------------------------------------------------------------------------------
 Limit  (cost=0.29..0.33 rows=1 width=16) (actual time=0.031..0.032 rows=1 loops=1)
   ->  LockRows  (cost=0.29..4.31 rows=100 width=16) (actual time=0.030..0.031 rows=1 loops=1)
         ->  Index Scan using idx_task_queue_pickup on task_queue
               (cost=0.29..4.31 rows=100 width=16) (actual time=0.014..0.024 rows=1 loops=1)
               Index Cond: (queue = 'default'::text)
 Planning Time: 0.108 ms
 Execution Time: 0.048 ms
sql
-- 差：没有索引，全表扫描 + 逐行加锁
DROP INDEX idx_task_queue_pickup;

EXPLAIN (ANALYZE, BUFFERS)
SELECT id FROM task_queue WHERE status = 'pending' LIMIT 1 FOR UPDATE SKIP LOCKED;
text
                              QUERY PLAN
------------------------------------------------------------------------
 Limit  (cost=0.00..1234.00 rows=1 width=16) (actual time=8.42..8.42 rows=1 loops=1)
   ->  LockRows  (cost=0.00..1234000.00 rows=10000 width=16) (actual time=8.41..8.41 rows=1 loops=1)
         ->  Seq Scan on task_queue  (cost=0.00..1234000.00 rows=10000 ...)
 Execution Time: 8.510 ms

部分索引是队列的标配

WHERE status = 'pending' 的部分索引（Partial Index）让索引只包含「待处理」的行。随着任务被消费，索引会自动变小，扫描成本随之下降——这对「先入先出」的队列尤其重要。

实战示例 ​

下面完整搭建一个多 worker 的任务队列，并用压测验证「无重复、无阻塞」。

第 1 步：建表与索引 ​

sql
DROP TABLE IF EXISTS task_queue;

CREATE TABLE task_queue (
  id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  queue        text NOT NULL DEFAULT 'default',
  payload      jsonb NOT NULL,
  status       text NOT NULL DEFAULT 'pending'
               CHECK (status IN ('pending', 'running', 'done', 'failed')),
  attempts     int NOT NULL DEFAULT 0,
  max_attempts int NOT NULL DEFAULT 3,
  locked_by    text,
  locked_at    timestamptz,
  run_at       timestamptz NOT NULL DEFAULT now(),
  created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX idx_task_queue_pickup
  ON task_queue (queue, run_at, id)
  WHERE status = 'pending';

ANALYZE task_queue;
text
DROP TABLE
CREATE TABLE
CREATE INDEX
ANALYZE

第 2 步：灌入 1 万条任务 ​

sql
INSERT INTO task_queue (queue, payload)
SELECT 'default', jsonb_build_object('seq', g, 'job', 'noop')
FROM generate_series(1, 10000) AS g;

SELECT count(*) AS pending, min(id) AS first_id, max(id) AS last_id
FROM task_queue
WHERE status = 'pending';
text
INSERT 0 10000
 pending | first_id | last_id
---------+----------+---------
   10000 |        1 |   10000

第 3 步：验证「并发抢占不重复」 ​

在会话 A 中抢占第一条：

sql
BEGIN;
SELECT id FROM task_queue
WHERE status = 'pending' AND queue = 'default'
ORDER BY run_at, id
LIMIT 1
FOR UPDATE SKIP LOCKED;
text
 id
----
  1
(1 row)

保持事务不提交，在会话 B 中执行同样的语句：

sql
SELECT id FROM task_queue
WHERE status = 'pending' AND queue = 'default'
ORDER BY run_at, id
LIMIT 1
FOR UPDATE SKIP LOCKED;
text
 id
----
  2
(1 row)

-- 会话 B 立刻返回了 2，没有等待，这正是队列所需的行为

第 4 步：验证「全部被锁时返回 0 行」 ​

会话 A 先锁住所有 pending 的行（SELECT id FROM task_queue WHERE status = 'pending' FOR UPDATE SKIP LOCKED;，返回 10000 行）并保持不提交；此时在会话 B 执行同一条语句：

text
 id
----
(0 rows)

返回 0 行而不是阻塞——消费者据此就知道「当前没有可处理的任务」，可以退避后重试。清理现场：两个会话都执行 ROLLBACK;。

第 5 步：完整的「抢占 + 处理 + 提交」事务 ​

sql
-- 模拟一个 worker 的完整生命周期
BEGIN;

-- 1) 原子抢占一条任务
WITH claimed AS (
  UPDATE task_queue
  SET status    = 'running',
      locked_by = 'worker-1',
      locked_at = now(),
      attempts  = attempts + 1
  WHERE id = (
    SELECT id FROM task_queue
    WHERE status = 'pending' AND queue = 'default' AND run_at <= now()
    ORDER BY run_at, id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
  )
  RETURNING id, payload, attempts
)
-- 2) 在同一事务里处理（这里是直接标记完成）
UPDATE task_queue t
SET status = 'done', locked_by = NULL, locked_at = NULL
FROM claimed c
WHERE t.id = c.id
RETURNING t.id, c.attempts;
text
  id  | attempts
------+----------
    1 |        1
(1 row)
sql
-- 3) 记录进度并提交
SELECT count(*) FILTER (WHERE status = 'done')    AS done,
       count(*) FILTER (WHERE status = 'pending') AS pending
FROM task_queue;
COMMIT;
text
 done | pending
------+---------
    1 |    9999
COMMIT

第 6 步：并发压测——验证无重复消费 ​

用 pgbench 或简单脚本并发调用下面的语句，然后检查是否有任务被重复领取：

sql
-- 每个 worker 反复执行这句
UPDATE task_queue
SET status    = 'running',
    locked_by = :'worker',
    locked_at = now(),
    attempts  = attempts + 1
WHERE id = (
  SELECT id FROM task_queue
  WHERE status = 'pending' AND queue = 'default'
  ORDER BY run_at, id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
RETURNING id;

-- 压测后的两项一致性验证
-- 验证 1：每个任务的 attempts 不应大于实际被领取次数
SELECT id, attempts, status FROM task_queue WHERE attempts > 1 ORDER BY id LIMIT 5;

-- 验证 2：running 状态的任务都应带有 locked_by，且不重复
SELECT locked_by, count(*) AS running_count
FROM task_queue
WHERE status = 'running'
GROUP BY locked_by
ORDER BY running_count DESC;
text
 id | attempts | status
----+----------+--------
(0 rows)

 locked_by | running_count
-----------+---------------
(0 rows)

第 7 步：验证「无死锁、无长等待」 ​

sql
-- 压测期间检查是否有锁等待堆积（应为空）
SELECT pid,
       pg_blocking_pids(pid) AS blocked_by,
       wait_event_type,
       wait_event,
       now() - query_start AS waited
FROM pg_stat_activity
WHERE wait_event_type = 'Lock';
text
 pid | blocked_by | wait_event_type | wait_event | waited
-----+------------+-----------------+------------+--------
(0 rows)

结果应该是空的——SKIP LOCKED 的设计目标就是「消费者之间永不阻塞」。

第 8 步：僵死任务回收 ​

sql
-- 回收 worker 崩溃后遗留的 running 任务
UPDATE task_queue
SET status    = 'pending',
    locked_by = NULL,
    locked_at = NULL
WHERE status = 'running'
  AND locked_at < now() - interval '10 minutes'
RETURNING id, locked_by, locked_at;
text
 id | locked_by | locked_at
----+-----------+-----------
(0 rows)

把这个回收逻辑做成定时任务（pg_cron 或应用侧调度），即可解决「worker 崩溃导致任务永久卡住」的问题。

注意事项 ​

错误！把 SKIP LOCKED 用在统计、对账、报表上。

sql
-- 错误！结果会静默漏掉被锁住的行，且每次执行结果不同
SELECT count(*) FROM orders FOR UPDATE SKIP LOCKED;
sql
-- 正确做法：统计用普通查询，不要锁行
SELECT count(*) FROM orders;

错误！在 UPDATE 语句上直接写 LIMIT / FOR UPDATE SKIP LOCKED。

sql
-- 错误！UPDATE 不支持 LIMIT，也不支持 FOR UPDATE 子句
UPDATE tasks SET status = 'running'
WHERE status = 'pending'
LIMIT 1
FOR UPDATE SKIP LOCKED;
text
ERROR:  syntax error at or near "LIMIT"
sql
-- 正确做法：用子查询承载 LIMIT + FOR UPDATE SKIP LOCKED
UPDATE tasks SET status = 'running'
WHERE id = (
  SELECT id FROM tasks WHERE status = 'pending'
  ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED
);

错误！先抢占提交，再慢慢处理。

sql
-- 错误！抢占后立刻提交，worker 崩溃则任务永久卡在 running
BEGIN;
UPDATE tasks SET status='running' WHERE id = 1;
COMMIT;
-- ... 在这里崩溃 → 任务再也回不到 pending（除非有回收机制）
sql
-- 正确做法：抢占与处理在同一事务内完成；或必须配合僵死回收
BEGIN;
UPDATE tasks SET status = 'running' WHERE id = 1;
-- ... 在事务内完成业务处理
UPDATE tasks SET status = 'done' WHERE id = 1;
COMMIT;

错误！以为 NOWAIT 和 SKIP LOCKED 可以组合使用。

sql
-- 错误！二者互斥
SELECT * FROM tasks FOR UPDATE NOWAIT SKIP LOCKED;
text
ERROR:  conflicting lock modes specified

不推荐：没有索引就上 SKIP LOCKED。 无索引时 PostgreSQL 要扫描全部候选行、逐行尝试加锁并检查跳过，队列一长就会退化成全表扫描 + 大量无效加锁，吞吐骤降。部分索引（WHERE status = 'pending'）是这类队列的标配。

最佳实践 ​

  • 用「子查询 + LIMIT 1 + FOR UPDATE SKIP LOCKED」作为抢占任务的标准写法，保证原子性：
sql
UPDATE task_queue
SET status = 'running', locked_by = :worker, locked_at = now()
WHERE id = (
  SELECT id FROM task_queue
  WHERE status = 'pending' AND queue = :queue
  ORDER BY run_at, id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
RETURNING id, payload;
  • 给队列建部分索引，让抢占保持常数级开销：
sql
CREATE INDEX idx_task_queue_pickup
  ON task_queue (queue, run_at, id)
  WHERE status = 'pending';
  • 为 SKIP LOCKED 加 lock_timeout 保护，避免异常情况下无限等待（虽然理论上不该发生）：
sql
SET lock_timeout = '2s';
  • 必须有僵死任务回收机制，否则 worker 崩溃会永久丢任务：
sql
UPDATE task_queue
SET status = 'pending', locked_by = NULL, locked_at = NULL
WHERE status = 'running' AND locked_at < now() - interval '10 minutes';
  • 失败重试用 attempts + max_attempts 控制，并做退避：
sql
UPDATE task_queue
SET status = CASE WHEN attempts >= max_attempts THEN 'failed' ELSE 'pending' END,
    run_at = now() + (interval '1 second' * power(2, attempts)),
    locked_by = NULL, locked_at = NULL
WHERE id = 42;
  • 队列场景使用 READ COMMITTED，不要用 SERIALIZABLE：
sql
SET default_transaction_isolation = 'read committed';
  • 消费者拿不到任务时要退避，避免空转打爆数据库：
sql
-- 返回 0 行后，应用应 sleep 几十到几百毫秒再重试
SELECT pg_sleep(0.2);
  • 只在语义允许跳过的场景使用 SKIP LOCKED，把它当作「抢占工具」而不是「查询优化器选项」：
text
适合：任务队列、多 worker 分摊、分片批处理、抢占式资源分配
禁止：统计、对账、报表、分页展示、全量迁移、唯一性校验

下一节

继续学习：死锁产生原因

参考链接 ​