用 PostgreSQL SKIP LOCKED 实现不阻塞的任务队列

问题场景

在服务端开发中，常见需求是把耗时任务（发邮件、生成报表、调用外部接口）放入队列，由多个 worker 并发消费。用消息中间件当然可以，但如果系统已经依赖 PostgreSQL，且任务量不大、不想引入额外组件，可以直接用表作为队列。

朴素做法是 SELECT ... WHERE status='pending' LIMIT 1，然后 UPDATE ... SET status='running'。两个 worker 可能同时读到同一行，导致重复处理。加 FOR UPDATE 可以锁行，但第二个 worker 会阻塞等待，吞吐下降。

PostgreSQL 9.5+ 提供 FOR UPDATE SKIP LOCKED：在锁定行时跳过已被其他事务锁定的行，让多个 worker 各自拿到不同的任务，互不阻塞。

表结构设计

CREATE TABLE jobs (
    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,
    max_attempts INT NOT NULL DEFAULT 5,
    run_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    locked_at   TIMESTAMPTZ,
    locked_by   TEXT,
    last_error  TEXT,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    CONSTRAINT jobs_status_check CHECK (status IN ('pending','running','done','failed'))
);

CREATE INDEX jobs_pending_idx
    ON jobs (run_at, id)
    WHERE status = 'pending';

部分索引只覆盖待处理任务，避免全表索引膨胀。run_at 用于延迟任务和重试退避。

原子拉取任务

一次 SQL 完成“选任务 + 标记 running + attempts+1”，避免两步之间被其他 worker 插入。

WITH picked AS (
    SELECT id
    FROM jobs
    WHERE status = 'pending'
      AND run_at <= now()
      AND attempts < max_attempts
    ORDER BY run_at, id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE jobs
SET status = 'running',
    locked_at = now(),
    locked_by = $1,
    attempts = jobs.attempts + 1,
    updated_at = now()
FROM picked
WHERE jobs.id = picked.id
RETURNING jobs.*;

参数 $1 是 worker 标识（如主机名 + 进程号，占位值）。注意 LIMIT 可调，一次拉多个减少往返。

在 READ COMMITTED 隔离级别下，FOR UPDATE SKIP LOCKED 会跳过被其他未提交事务锁定的行；提交后这些行对当前查询不可见，因此不会重复。单条语句在自动提交模式下自身就是一个事务，锁在语句结束前有效；若拆成两条语句，必须显式包在 BEGIN ... COMMIT 中。

Python worker 最小示例

依赖：psycopg2-binary。假设连接串从环境变量读取，示例用占位值。

import os, time, socket
import psycopg2

DSN = os.environ.get('DATABASE_URL', 'postgresql://user:pass@localhost:5432/mydb')
WORKER_ID = f'{socket.gethostname()}-{os.getpid()}'

PULL_SQL = '''
WITH picked AS (
    SELECT id FROM jobs
    WHERE status = 'pending'
      AND run_at <= now()
      AND attempts < max_attempts
    ORDER BY run_at, id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE jobs
SET status = 'running',
    locked_at = now(),
    locked_by = %(worker)s,
    attempts = attempts + 1,
    updated_at = now()
FROM picked
WHERE jobs.id = picked.id
RETURNING jobs.*;
'''

def handle(job):
    # 占位业务：只打印，便于演示
    print('处理任务', job['id'], job['kind'], job['payload'])

def main():
    conn = psycopg2.connect(DSN)
    conn.autocommit = True
    while True:
        with conn.cursor() as cur:
            cur.execute(PULL_SQL, {'worker': WORKER_ID})
            row = cur.fetchone()
            cols = [d[0] for d in cur.description]
        if not row:
            time.sleep(1)
            continue
        job = dict(zip(cols, row))
        try:
            handle(job)
            with conn.cursor() as cur:
                cur.execute(
                    '''UPDATE jobs SET status='done', locked_at=NULL,
                       locked_by=NULL, updated_at=now() WHERE id=%s''',
                    (job['id'],),
                )
        except Exception as e:
            backoff = min(60, 2 ** job['attempts'])
            with conn.cursor() as cur:
                if job['attempts'] >= job['max_attempts']:
                    cur.execute(
                        '''UPDATE jobs SET status='failed', last_error=%s,
                           locked_at=NULL, locked_by=NULL, updated_at=now()
                           WHERE id=%s''',
                        (str(e)[:500], job['id']),
                    )
                else:
                    cur.execute(
                        '''UPDATE jobs SET status='pending', last_error=%s,
                           run_at = now() + (%s || ' seconds')::interval,
                           locked_at=NULL, locked_by=NULL, updated_at=now()
                           WHERE id=%s''',
                        (str(e)[:500], backoff, job['id']),
                    )

if __name__ == '__main__':
    main()

运行方法：

1. 在 PostgreSQL 中执行表结构和索引 SQL。
2. 安装依赖：pip install psycopg2-binary。
3. 设置 DATABASE_URL，启动多个进程：python worker.py。可以开多个终端或使用 &。
4. 插入测试任务：

INSERT INTO jobs (kind, payload) VALUES ('demo', jsonb_build_object('n', 1)), ('demo', jsonb_build_object('n', 2));

观察每个任务只被一个 worker 处理。

重试与超时恢复

上面的异常分支把任务改回 pending 并设置 run_at 延迟，attempts 已经在拉取时增加，因此不会无限重试。当 attempts >= max_attempts 标记 failed。

如果 worker 进程崩溃，任务会停留在 running。需要一个恢复查询，由独立定时任务执行：

UPDATE jobs
SET status = 'pending',
    locked_at = NULL,
    locked_by = NULL,
    run_at = now(),
    updated_at = now()
WHERE status = 'running'
  AND locked_at < now() - interval '10 minutes';

超时阈值应大于单任务最长执行时间。由于崩溃恢复和正常重试都可能让同一任务再次执行，整体语义是“至少一次”，业务处理应尽量幂等，或依赖业务唯一键去重。

边界与限制

• SKIP LOCKED 需要 PostgreSQL 9.5 及以上。旧版本不支持。
• 该方案适合中小规模、任务执行时间相对可控的场景。任务量极大时，单表膨胀、索引维护和锁竞争会成为瓶颈，此时应考虑专用消息系统。
• 长事务会持有行锁，拖慢其他 worker。拉取后应尽快提交状态变更；不要在处理业务逻辑期间持有未提交事务。
• FOR UPDATE 与 ORDER BY ... LIMIT 组合时，SKIP LOCKED 跳过的行在本次查询中不会返回，但下次查询可能拿到。排序只保证可见行的顺序，不保证全局严格顺序。
• 连接池中每个 worker 长期占用一个连接。worker 很多时注意连接数上限。
• locked_by 只用于排查，不能作为抢占依据；超时判断只用 locked_at。
• run_at 使用数据库时间 now()，避免各 worker 机器时钟偏差。
• 删除已完成任务要定期做，否则表会持续增长；可以按 updated_at 分批删除。

总结

SELECT ... FOR UPDATE SKIP LOCKED 把行锁和跳过机制结合起来，让多个 worker 并发领取不同任务而不互相阻塞。配合 status、attempts、max_attempts、run_at 和 locked_at，可以实现原子拉取、延迟重试和崩溃恢复。它不能替代消息中间件，但在已有 PostgreSQL 的项目里，是一个足够简单且行为可解释的队列方案。