FOR UPDATE SKIP LOCKEDでジョブキュー｜PostgreSQL実測

ジョブの状態は queued → running → done / failed の一方通行にし、失敗と「置き去り」だけが queued に戻る形にします。この記事のコードと実測は、すべてこの状態遷移の上に載っています。

flowchart TD
    Q[queued] -->|取得| R[running]
    R -->|成功| D[done]
    R -->|失敗<br>再試行あり| Q
    R -->|失敗<br>上限に到達| F[failed]
    R -.->|放置された<br>runningを回収| Q

先に、調べられることの多い点の答えをまとめます。

確認環境：Ubuntu 24.04.5 LTS（2 CPU・メモリ 8GB の共用マシン）の上で、PostgreSQL 16.15（Ubuntu パッケージ）を自分用のデータディレクトリに initdb し、127.0.0.1 の専用ポートで起動しました（log_lock_waits=on・deadlock_timeout=200ms 以外は既定値）。2セッションの実演は psql 16.15、ワーカーと処理量の計測は Python 3.13.18＋psycopg 3.3.7（libpq 18）を multiprocessing で 1〜8 プロセス起動して行い、二重取得は「取得したジョブ ID を別テーブルに記録して GROUP BY で数える」方法で数えました。比較用の Redis 7.0.15 は永続化を切って 127.0.0.1 に起動し、redis-py 8.1.0 で LMOVE と Streams（XREADGROUP）を使いました。この記事で実際に動かしたものは、上記の PostgreSQL・psql・Python ワーカー・Redis のすべてで、SQS・MySQL・Oracle の挙動は公式ドキュメントの範囲です。処理量の数値は 2 CPU を PostgreSQL とワーカーが分け合う小さな環境のもので、桁の目安として読む値です（確認日 2026年11月4日）。

FOR UPDATE・NOWAIT・SKIP LOCKED の違い（2セッションで実演）

この節では、同じ行をロックしている別セッションがいるとき、3つの書き方がそれぞれどう振る舞うかが、psql の原文で分かります。

違いは「ロック済みの行に当たったときの動き」だけです。句が無ければコミットまで待ち、NOWAIT は即座にエラー、SKIP LOCKED はその行を無かったことにして次へ進みます。公式ドキュメント（SELECT のロック句）は SKIP LOCKED について「ロックされた行を飛ばすとデータの一貫しない見え方になるので汎用の用途には向かないが、複数のコンシューマがキューのようなテーブルにアクセスするときのロック競合を避けるために使える」と書いています。つまりキューのための機能です。

セッション A が id 1・2 を FOR UPDATE で押さえたまま 6 秒眠り、その間にセッション B が3通りの SELECT を順に流しました。

-- セッション A
BEGIN;
SELECT id, status FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 2 FOR UPDATE;
 id | status
----+--------
  1 | queued
  2 | queued
SELECT pg_sleep(6);
COMMIT;

-- セッション B（\timing on）
SELECT id, status FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 2 FOR UPDATE NOWAIT;
ERROR:  55P03: could not obtain lock on row in relation "jobs"
LOCATION:  heap_lock_tuple, heapam.c:4832
Time: 1.485 ms

SELECT id, status FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 2 FOR UPDATE SKIP LOCKED;
 id | status
----+--------
  3 | queued
  4 | queued
Time: 1.520 ms

SELECT id, status FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 2 FOR UPDATE;
 id | status
----+--------
  1 | queued
  2 | queued
Time: 5019.987 ms (00:05.020)

3つ目の素の FOR UPDATE は、A のコミットまで 5 秒待ってから、A が押さえていた 1・2 を受け取りました。log_lock_waits=on にしておくと、サーバーログにも待ちの記録が残ります。

LOG:  process 2798 still waiting for ShareLock on transaction 736 after 201.172 ms
LOG:  process 2798 acquired ShareLock on transaction 736 after 5019.278 ms

待っている間に何が起きているか（pg_locks）

B が待っている最中に、3つ目のセッションから pg_locks と pg_blocking_pids() を見ました。A（pid 3397）はトランザクション ID 739 の ExclusiveLock を持ち、B（pid 3400）はその 739 に対する ShareLock を granted = f で待っています。B がページ 0・タプル 1 の tuple ロックを持っているのは「その行の次の順番は私」という席取りで、行ロックの待ち行列はこうして作られます。

 pid  |   locktype    | rel  | page | tuple | xid |        mode         | granted
------+---------------+------+------+-------+-----+---------------------+---------
 3397 | relation      | jobs |      |       |     | RowShareLock        | t
 3397 | relation      | jobs |      |       |     | RowExclusiveLock    | t
 3397 | transactionid |      |      |       | 739 | ExclusiveLock       | t
 3400 | relation      | jobs |      |       |     | RowShareLock        | t
 3400 | transactionid |      |      |       | 739 | ShareLock           | f
 3400 | tuple         | jobs |    0 |     1 |     | AccessExclusiveLock | t

 pid  | state  | wait_event_type |  wait_event   | query
------+--------+-----------------+---------------+------------------------------------------
 3400 | active | Lock            | transactionid | SELECT id, status FROM jobs WHERE ...

 pid  | blocked_by
------+------------
 3400 | {3397}

このときの A は、取った行を status = 'running' に更新してからコミットしました。B の素の FOR UPDATE は 3.3 秒待ったあと、id 1 ではなく id 2 を返しました。READ COMMITTED では、待ち明けに行を再評価して WHERE（status = 'queued'）を満たさなくなった行を捨て、次の行を探しに行くためです。つまり待ち時間はまるごと無駄になります。SKIP LOCKED なら同じ結果を 0.7 ミリ秒で得ました。ロック中の行の調べ方は関連記事「PostgreSQLロック確認SQL｜ブロッキング特定から解除までコピペ集」で扱っています。

FOR UPDATE と FOR NO KEY UPDATE（外部キーがあるときの落とし穴）

行ロックには強さが4段階あり、公式ドキュメント（13.3.2 行レベルロック）の競合表では、FOR UPDATE だけが FOR KEY SHARE と競合します。FOR KEY SHARE は、外部キーで jobs を参照する子テーブルに INSERT するとき、親行に対して自動的に取られるロックです。手元で job_logs (job_id REFERENCES jobs) を作り、セッション A が id 1 をロックしている間に別セッションから進捗ログを INSERT しました。

ジョブの取得で主キーを書き換えたり行を削除したりすることは無いので、子テーブルがある設計では FOR NO KEY UPDATE SKIP LOCKED のほうが待ちを作りません。なお UPDATE jobs SET status = … 自体が取るのも NO KEY UPDATE の強さです。この記事の以降のコードは、検索語に合わせて FOR UPDATE で書いていますが、読み替えて差し支えありません。

ジョブテーブルの設計（状態・列・部分インデックス）

この節では、jobs テーブルの DDL と、各列が「どの設計判断のために要るか」が分かります。

列は「取得条件」「再試行」「回収」「調査」の4つの目的で決めます。この記事で使った DDL は次のとおりです。

CREATE TYPE job_status AS ENUM ('queued', 'running', 'done', 'failed');

CREATE TABLE jobs (
  id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  queue        text        NOT NULL DEFAULT 'default',
  payload      jsonb       NOT NULL,
  status       job_status  NOT NULL DEFAULT 'queued',
  priority     int         NOT NULL DEFAULT 0,
  run_at       timestamptz NOT NULL DEFAULT now(),   -- この時刻以降に実行してよい
  attempts     int         NOT NULL DEFAULT 0,
  max_attempts int         NOT NULL DEFAULT 3,
  locked_by    text,                                 -- 取ったワーカー名
  locked_at    timestamptz,                          -- 取った時刻（回収の判定に使う）
  last_error   text,
  created_at   timestamptz NOT NULL DEFAULT now(),
  finished_at  timestamptz
);

-- 取得クエリの WHERE と ORDER BY にそのまま対応する部分インデックス
CREATE INDEX jobs_dequeue_idx ON jobs (queue, priority DESC, run_at, id) WHERE status = 'queued';
-- 置き去りの running を探すための部分インデックス
CREATE INDEX jobs_running_idx ON jobs (locked_at) WHERE status = 'running';

取得クエリの書き方（UPDATE … FOR UPDATE SKIP LOCKED … RETURNING）

この節では、取得を1文で済ませる書き方と、ORDER BY・LIMIT・CTE に関する公式ドキュメントの注意点が分かります。

SELECT で行をロックし、同じ文の UPDATE で running に変え、RETURNING で中身を受け取ります。SELECT と UPDATE を別の文にすると、間に他のワーカーが割り込む余地と、往復が1回増えます。

UPDATE jobs
   SET status = 'running', locked_by = $1, locked_at = now(), attempts = attempts + 1
 WHERE id IN (SELECT id FROM jobs
               WHERE queue = $2 AND status = 'queued' AND run_at <= now()
               ORDER BY priority DESC, run_at, id
               LIMIT $3
               FOR UPDATE SKIP LOCKED)
RETURNING id, payload, attempts, max_attempts;

同じことを CTE で書く形もよく見ます。公式ドキュメントには「ロック句は主問い合わせから参照される WITH 問い合わせには適用されない。WITH の中で行ロックをかけたいなら WITH の中にロック句を書く」とあるので、ロック句は必ず WITH の内側に置きます。

WITH next_jobs AS (
  SELECT id FROM jobs
   WHERE queue = $2 AND status = 'queued' AND run_at <= now()
   ORDER BY priority DESC, run_at, id
   LIMIT $3
   FOR UPDATE SKIP LOCKED
)
UPDATE jobs j SET status = 'running', locked_by = $1, locked_at = now(), attempts = attempts + 1
  FROM next_jobs n WHERE j.id = n.id
RETURNING j.id, j.payload, j.attempts, j.max_attempts;

順序について公式ドキュメントは「READ COMMITTED で ORDER BY とロック句を併用すると、待ちの間に順序列が更新された場合に順序が崩れて見えることがある」と注意しています。ジョブキューでは「おおむね優先順」で足りることがほとんどで、厳密な順序が要る処理（口座ごとの取引の逐次処理など）は、この方式ではなく対象ごとの直列化を別に設計します。

EXPLAIN で見るロックの取り方（LockRows と部分インデックス）

この節では、取得クエリの実行計画のどこでロックが取られ、部分インデックスが無いと何が起きるかが分かります。

実行計画では LockRows ノードが Limit の下に置かれ、スキャンした行を1行ずつロックして、取れた行だけを上に渡します。SKIP LOCKED で飛ばした行は LIMIT の数に数えられないので、「10 件頼んだのにロック中の行のせいで 7 件しか来ない」ことは起きません。100,000 件を入れた状態での計画です。

EXPLAIN (ANALYZE, BUFFERS, COSTS OFF)
SELECT id FROM jobs WHERE queue = 'default' AND status = 'queued' AND run_at <= now()
 ORDER BY priority DESC, run_at, id LIMIT 10 FOR UPDATE SKIP LOCKED;

 Limit (actual time=0.048..0.060 rows=10 loops=1)
   Buffers: shared hit=14
   ->  LockRows (actual time=0.047..0.058 rows=10 loops=1)
         Buffers: shared hit=14
         ->  Index Scan using jobs_dequeue_idx on jobs (actual time=0.034..0.039 rows=10 loops=1)
               Index Cond: ((queue = 'default'::text) AND (run_at <= now()))
               Filter: (status = 'queued'::job_status)
               Buffers: shared hit=4
 Execution Time: 0.090 ms

Index Scan が 4 ブロック、ロックで 10 ブロック（行をロックするとタプルヘッダーに書き込むので、ヒープのページを触ります）。部分インデックスを DROP して同じクエリを流すと、全行を読んでディスク上でソートしてから先頭 10 件をロックする計画になり、0.09 ms が 71 ms になりました。

 Limit (actual time=70.635..70.644 rows=10 loops=1)
   Buffers: shared hit=1448, temp read=480 written=491
   ->  LockRows (actual time=70.634..70.641 rows=10 loops=1)
         ->  Sort (actual time=70.620..70.622 rows=10 loops=1)
               Sort Key: priority DESC, run_at, id
               Sort Method: external merge  Disk: 3920kB
               ->  Seq Scan on jobs (actual time=0.012..18.512 rows=99990 loops=1)
                     Filter: ((queue = 'default'::text) AND (status = 'queued'::job_status) AND (run_at <= now()))
 Execution Time: 71.027 ms

UPDATE 全体の計画は、サブクエリの LockRows の結果を HashAggregate で重複排除し、主キーで本体の行を引いて更新する形でした（Execution Time 0.421 ms）。設計のうえで押さえる点は、LockRows が Limit の内側にあること、その下が Index Scan（部分インデックス）になっていること、の 2 つです。

Python（psycopg）ワーカーの完成コード

この節では、取得→処理→done／再試行／failed、置き去りの回収、SIGTERM での停止まで入ったワーカーのコードが分かります。

ワーカーは autocommit の接続で「1文ずつ確定」させ、処理の間はトランザクションを張りません。処理の間ロックを持ち続ける設計との比較は「置き去りになった running の回収」の節で扱います。

#!/usr/bin/env python3
"""SKIP LOCKED ジョブキューのワーカー（psycopg 3）
   取得(queued→running) → 処理 → done / 再試行(queued, run_at を後ろへ) / failed"""
import os, signal, time
import psycopg

DSN = os.environ.get("DSN", "host=127.0.0.1 port=54241 user=postgres dbname=t241db")
WORKER = os.environ.get("WORKER", f"worker-{os.getpid()}")
BATCH = int(os.environ.get("BATCH", "1"))
POLL_SEC = float(os.environ.get("POLL_SEC", "0.5"))
STALE_SEC = int(os.environ.get("STALE_SEC", "60"))

CLAIM_SQL = """
UPDATE jobs
   SET status = 'running', locked_by = %(w)s, locked_at = now(), attempts = attempts + 1
 WHERE id IN (SELECT id FROM jobs
               WHERE queue = %(q)s AND status = 'queued' AND run_at <= now()
               ORDER BY priority DESC, run_at, id
               LIMIT %(n)s
               FOR UPDATE SKIP LOCKED)
RETURNING id, payload, attempts, max_attempts
"""
DONE_SQL = """UPDATE jobs SET status = 'done', finished_at = now(), locked_by = NULL, locked_at = NULL
               WHERE id = %s AND status = 'running'"""
RETRY_SQL = """
UPDATE jobs
   SET status = CASE WHEN attempts >= max_attempts THEN 'failed'::job_status ELSE 'queued'::job_status END,
       run_at = now() + make_interval(secs => least(300, 2 ^ attempts)),   -- 指数バックオフ（上限300秒）
       last_error = left(%s, 2000),
       finished_at = CASE WHEN attempts >= max_attempts THEN now() END,
       locked_by = NULL, locked_at = NULL
 WHERE id = %s AND status = 'running'
RETURNING status, run_at
"""
REAP_SQL = """
UPDATE jobs
   SET status = CASE WHEN attempts >= max_attempts THEN 'failed'::job_status ELSE 'queued'::job_status END,
       last_error = concat_ws(E'\\n', last_error, 'stale running: ' || locked_by),
       locked_by = NULL, locked_at = NULL
 WHERE status = 'running' AND locked_at < now() - make_interval(secs => %s)
"""


def handle(payload: dict) -> None:
    """ここが実際の処理。例として n が 7 の倍数なら失敗する"""
    if payload.get("n", 0) % 7 == 0:
        raise RuntimeError(f"n={payload['n']} は 7 の倍数なので失敗")
    time.sleep(0.01)


def main():
    stop = False
    def on_term(*_):
        nonlocal stop; stop = True
    signal.signal(signal.SIGTERM, on_term); signal.signal(signal.SIGINT, on_term)
    with psycopg.connect(DSN, autocommit=True, application_name=WORKER) as conn:
        last_reap = 0
        while not stop:
            if time.time() - last_reap > 10:            # 10秒に1回、置き去りの running を回収
                conn.execute(REAP_SQL, (STALE_SEC,)); last_reap = time.time()
            rows = conn.execute(CLAIM_SQL, {"w": WORKER, "q": "default", "n": BATCH}).fetchall()
            if not rows:
                time.sleep(POLL_SEC); continue
            for job_id, payload, attempts, max_attempts in rows:
                try:
                    handle(payload)
                    conn.execute(DONE_SQL, (job_id,))
                    print(f"{WORKER} job={job_id} done (attempt {attempts})", flush=True)
                except Exception as e:
                    st, run_at = conn.execute(RETRY_SQL, (f"{type(e).__name__}: {e}", job_id)).fetchone()
                    print(f"{WORKER} job={job_id} {st} (attempt {attempts}/{max_attempts}) next={run_at:%H:%M:%S}", flush=True)


if __name__ == "__main__":
    main()

10 件のジョブを入れ、ワーカーを 2 つ起動して 12 秒後に SIGTERM で止めたログです。n = 7 のジョブだけ失敗し、2 秒・4 秒・8 秒と間隔を空けて 3 回目で failed になりました。

worker-2 job=1 done (attempt 1)
worker-1 job=2 done (attempt 1)
worker-2 job=3 done (attempt 1)
worker-1 job=4 done (attempt 1)
worker-2 job=5 done (attempt 1)
worker-1 job=6 done (attempt 1)
worker-2 job=7 queued (attempt 1/3) next=03:41:17
worker-2 job=8 done (attempt 1)
worker-1 job=9 done (attempt 1)
worker-2 job=10 done (attempt 1)
worker-1 job=7 queued (attempt 2/3) next=03:41:21
worker-1 job=7 failed (attempt 3/3) next=03:41:29

 id | status | attempts |               last_error                |  run_at
----+--------+----------+-----------------------------------------+----------
  6 | done   |        1 |                                         | 03:41:15
  7 | failed |        3 | RuntimeError: n=7 は 7 の倍数なので失敗 | 03:41:29
  8 | done   |        1 |                                         | 03:41:15

並列ワーカーの実測：二重取得 0 件と処理量（1〜8 並列・バッチ 1〜50 件）

この節では、ロック無し・FOR UPDATE・SKIP LOCKED の二重取得の件数と、並列数・バッチサイズで処理量がどう変わるかが分かります。

ロック無しの取得は 5,000 件中 3,616 件（72%）が 2 つ以上のワーカーに取られ、SKIP LOCKED と FOR UPDATE は 0 件でした。計測は、取得したジョブ ID をワーカー名とともに別テーブルへ INSERT し、終了後に GROUP BY job_id HAVING count(*) > 1 で数えています（この INSERT 1 回ぶんは本番のワーカーには無い計測用のオーバーヘッドです）。ロック無し版は「SELECT … LIMIT 1 で id を読む → UPDATE で running にする」の 2 文です。

ロック無しが遅いのは、同じジョブを複数のワーカーが処理して延べ回数が倍になるうえ、UPDATE 同士で行ロックの待ちが起きるためです。FOR UPDATE は正しいものの、全ワーカーが先頭の同じ行に並ぶので、ワーカーを増やしても待ちが増えるだけになります。

並列数とバッチサイズ

1 件あたりの処理時間が 0 のとき（取得と完了の往復だけ）と、5 ミリ秒の I/O 待ちを入れたときの処理量です。2 CPU を PostgreSQL と Python が分け合っているので、処理時間 0 では 4 並列を超えると頭打ちになります。

処理に I/O 待ちがあると、並列数にほぼ比例して伸びます（1 → 8 並列で 119 → 614 件/秒）。FOR UPDATE も処理時間があるうちは差が小さく、8 並列で 1 割ほど落ちる程度でした。差が開くのは「取得の競合が支配的になる」短いジョブのときです。

件数を 20,000 件に増やし、バッチサイズと synchronous_commit を変えた結果です。1 件ずつ取ると 1 ジョブあたり COMMIT が 2 回（取得と完了）あり、その都度 WAL の fsync が入ります。

バッチで取る効果が一番大きく、fsync を省く効果はその次です。synchronous_commit = off はクラッシュ時に直近のコミットを失う設定なので、ジョブキューには向きません（取ったはずのジョブが queued に戻る、done にしたジョブが running に戻る）。バッチサイズは「1 バッチの処理時間が回収の猶予（STALE_SEC）よりはるかに短い」範囲で決めます。

再試行と指数バックオフ（run_at・attempts・failed）

この節では、失敗したジョブをいつ・何回やり直し、どこで諦めるかの実装が分かります。

再試行は「queued に戻して run_at を未来にする」だけで実現でき、取得クエリの run_at <= now() がそのまま待ちになります。専用の列も別テーブルも要りません。先ほどのログの job 7 は、attempts = 1 で 2 秒後、2 で 4 秒後、3 で failed でした（2 ^ attempts 秒、上限 300 秒）。

failed になったジョブは消さずに残し、原因を直したあとに UPDATE jobs SET status = 'queued', attempts = 0, run_at = now() WHERE status = 'failed' AND last_error LIKE '%Timeout%' のように戻します。デッドレターキュー（DLQ）に相当するのがこの failed 行です。再試行の設計全体は関連記事「非同期処理の設計：同期との使い分け基準とジョブキューのリトライ設計」で扱っています。

置き去りになった running の回収（可視性タイムアウト相当）

この節では、ワーカーが処理中に落ちたときにジョブがどうなるかと、2つの回収の設計の違いが分かります。

autocommit で取得するワーカーが kill -9 で落ちると、そのジョブは running のまま残ります。誰も done にも failed にもしないので、locked_at の経過時間で見つけて戻すしかありません。これが SQS の可視性タイムアウト（受信したメッセージが一定時間後に再び見えるようになる仕組み。公式ドキュメントの既定は 30 秒）にあたります。

$ python crash_worker.py &
claimed (1,) pid 29840
$ kill -9 29840            # 処理中に落ちた想定

 id | status  | attempts | locked_by  |      age
----+---------+----------+------------+---------------
  1 | running |        1 | w-pid29840 | 00:00:01.6392
  2 | queued  |        0 |            |
  3 | queued  |        0 |            |

-- 回収（STALE_SEC を 60 秒とした場合。手元の実演では 0 秒で即時に実行）
UPDATE jobs
   SET status = CASE WHEN attempts >= max_attempts THEN 'failed'::job_status ELSE 'queued'::job_status END,
       run_at = now() + interval '5 seconds' * attempts,
       last_error = concat_ws(E'\n', last_error, 'stale running recovered at ' || now()::text),
       locked_by = NULL, locked_at = NULL
 WHERE status = 'running' AND locked_at < now() - interval '60 seconds'
RETURNING id, status, attempts, run_at;

 id | status | attempts |            run_at
----+--------+----------+-------------------------------
  1 | queued |        1 | 2026-11-04 03:40:21.859582+09

対案：処理の間トランザクションを張ったままにする

取得から完了までを1つのトランザクションにし、行ロックを持ったまま処理する設計もあります。ワーカーが落ちれば接続が切れてロールバックされ、行は queued のまま誰も触っていない状態に戻るので、回収の仕組みが要りません。ただし「接続が切れたことにサーバーが気づく」までは、ロックは残ります。手元で、トランザクションを張ったまま pg_sleep(60) している psql を kill -9 しました。

回収の猶予（STALE_SEC）は「正常なジョブの最大処理時間 × 2」程度から始め、超えそうな長いジョブは処理の途中で UPDATE jobs SET locked_at = now() WHERE id = $1 AND status = 'running' を打って延長します（SQS の ChangeMessageVisibility にあたります）。

ポーリング間隔と LISTEN/NOTIFY（空振りを減らす）

この節では、空のキューを叩き続けるコストと、INSERT をトリガーにワーカーを起こす方法が分かります。

空振りの SELECT は 0.05 ミリ秒前後と軽いので、0.5〜1 秒のポーリングは数十ワーカーまで問題になりません。それでも「投入から着手までの遅れを数ミリ秒にしたい」「ワーカーが多く、空振りの接続が無駄」というときは、INSERT のトリガーで pg_notify を送り、ワーカー側は LISTEN で寝て待ちます。

CREATE OR REPLACE FUNCTION jobs_notify() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  PERFORM pg_notify('jobs_' || NEW.queue, NEW.id::text);
  RETURN NEW;
END $$;
CREATE OR REPLACE TRIGGER jobs_notify_ins AFTER INSERT ON jobs FOR EACH ROW EXECUTE FUNCTION jobs_notify();

with psycopg.connect(DSN, autocommit=True) as c:
    c.execute("LISTEN jobs_default")
    for n in c.notifies(timeout=3):        # 通知が来るか 3 秒経つまでブロック
        print(f"通知: channel={n.channel} payload={n.payload} pid={n.pid}")
        # ここで CLAIM_SQL を実行する（通知の件数とジョブの件数は一致しない前提で）

LISTEN jobs_default ... 待機中
通知: channel=jobs_default payload=2 pid=32085

NOTIFY を送ってから Python 側で受け取るまでの遅れを 20 回測ると、中央値 0.52 ミリ秒・最大 0.69 ミリ秒でした。通知は「起こすため」だけに使い、取得は必ず SKIP LOCKED のクエリで行います（通知はコミット時にまとめて届き、同一ペイロードはまとめられるため、通知の数を信用しない）。また、通知を取りこぼした場合に備えて、LISTEN していても数秒に 1 回はポーリングを残します。

dead tuple と autovacuum（キューのテーブルが太る理由と対策）

この節では、ジョブテーブルがなぜ太りやすいのか、太るとどう遅くなるのか、どう抑えるかが実測で分かります。

1 ジョブで UPDATE が 2 回（running・done）走り、しかも status は部分インデックスの条件列なので HOT 更新になりません。50,000 件を 4 並列で流した直後の pg_stat_user_tables と pgstattuple です。

  ins  |  upd   | hot_upd | live  |  dead  | autovacuum_count
-------+--------+---------+-------+--------+------------------
 50000 | 100000 |       0 | 50000 | 100000 |                0

 table_size |  pkey   | dequeue_idx | total
------------+---------+-------------+-------
 10 MB      | 2200 kB | 3752 kB     | 17 MB

 leaf_pages | deleted_pages | avg_leaf_density     ← pgstatindex('jobs_dequeue_idx')
------------+---------------+------------------
        463 |             0 |            58.88

注目すべきは jobs_dequeue_idx です。部分インデックスなので「queued の行だけ」が入るはずですが、done に更新してもインデックスの不要になったエントリは VACUUM まで消えません。全件が done の状態でジョブを 1 件だけ投入し、取得クエリを流すと、インデックスの死んだエントリを 461 ブロックぶんたどってから目的の 1 行に着きました。VACUUM 後は 4 ブロックです。

VACUUM (VERBOSE) jobs;
INFO:  finished vacuuming "t241db.public.jobs": index scans: 1
pages: 0 removed, 1327 remain, 1327 scanned (100.00% of total)
tuples: 5752 removed, 50000 remain, 0 are dead but not yet removable
index scan needed: 1327 pages from table (100.00% of total) had 100000 dead item identifiers removed
index "jobs_pkey": pages: 275 in total, 0 newly deleted, 0 currently deleted, 0 reusable
index "jobs_dequeue_idx": pages: 469 in total, 465 newly deleted, 465 currently deleted, 0 reusable

469 ページ中 465 ページが「削除済み」になりました（ファイルの大きさは 3752 kB のまま。空いたページは次の挿入で再利用されます）。これが autovacuum に追いつかれないまま数百万件が流れると、取得クエリが数十ミリ秒〜秒単位に伸び、「SKIP LOCKED なのに遅い」という症状になります。

Redis・SQS・専用キューとの判断表（Redis は手元で比較）

この節では、同じジョブを Redis で流したときの処理量と、PostgreSQL のキューを選ぶ・選ばない判断の軸が分かります。

処理量だけなら Redis が 1 件ずつで 5 倍速く、PostgreSQL は「ジョブの投入と業務データの更新を同じトランザクションで確定できる」ことで選びます。永続化を切った Redis 7.0.15 と、fsync ありの PostgreSQL 16 を、同じ 4 並列・20,000 件・処理時間 0 で比べました（どちらも 1 ジョブあたり 3 コマンド／3 文）。

処理時間 5 ミリ秒・8 並列では、Redis List 1,350 件/秒、Redis Streams 1,339 件/秒に対して PostgreSQL は 614 件/秒でした。ジョブ 1 件の処理が数十ミリ秒以上あれば、どちらを選んでもキューがボトルネックにはなりません。

1. 1ジョブの投入が業務データの更新と同時に確定しないと困る → PostgreSQL注文確定と「確認メール送信ジョブ」を 1 トランザクションで。別キューだと Outbox パターンが要る
2. 2毎秒数千件以下・運用する部品を増やしたくない → PostgreSQLバックアップ・監視・権限が既存の DB と同じで済む。SQL で状態を調べられる
3. 3複数言語のワーカー・既存の Redis がある → Redis Streams か Celery/Sidekiq 等コンシューマグループ・XAUTOCLAIM・再試行の慣習が揃っている
4. 4AWS 上でワーカーの台数が増減する・配送保証をマネージドにしたい → SQS可視性タイムアウト・DLQ・メトリクスが標準。FIFO キューのスループット上限（高スループットモード以外は秒間 300 件、バッチで 3,000 件）は公式に明記
5. 5複数のサービスが同じイベントを読む・再生したい → Kafka 等のログ型「1 件を 1 ワーカーが取って消す」キューとは目的が違う

Redis の比較では List（LMOVE で処理中リストへ移す「信頼性キュー」）と Streams（コンシューマグループ）の 2 つの型だけを使い、どちらも二重取得は 0 件でした。Redis List は落ちたワーカーの処理中リストを自分で走査して戻す必要があり、PostgreSQL の locked_at による回収と手間は同じです。SQS・Kafka の挙動は公式ドキュメントの範囲で、手元では動かしていません。

よくあるエラー・つまずき（原文つき）

この節では、SKIP LOCKED のキューを組むときに実際に出るエラーと症状、その原因と直し方が一覧で分かります。

楽観ロックで組む案（version 列で UPDATE … WHERE version = $n を試み、0 行なら次へ）もありますが、ワーカーが多いと試行の無駄が増えます。2 つのロック方式の実測は関連記事「楽観ロックと悲観ロックの違い｜PostgreSQL実測でわかる使い分け」で扱っています。

よくある質問

Q. SELECT FOR UPDATE SKIP LOCKED とは何ですか？ A. SELECT で行ロックを取るとき、他のトランザクションがロック中の行を「待たずに飛ばす」指定です。PostgreSQL 9.5 で入りました。公式ドキュメントは、複数のコンシューマがキューのようなテーブルを読むときのロック競合を避けるための機能と説明しています。

Q. FOR UPDATE NOWAIT と SKIP LOCKED の違いは？ A. どちらも待ちません。NOWAIT はロック中の行に当たると could not obtain lock on row in relation（SQLSTATE 55P03）のエラーで文全体が失敗し、SKIP LOCKED はその行を除いた結果を返します。「その行でなければ困る」なら NOWAIT、「空いている行ならどれでも」なら SKIP LOCKED です。

Q. 本当に二重取得は起きませんか？ A. 手元では 1〜8 並列・5,000〜50,000 件のすべての回で 0 件でした。ロックは行単位でトランザクションの終わりまで保たれ、UPDATE で status が running になった行は他のワーカーの WHERE から外れます。二重に「処理」される可能性が残るのは、置き去りの回収で queued に戻した後に元のワーカーが生きていた場合で、処理を冪等に書くことで備えます。

Q. ORDER BY と LIMIT を付けても順序は守られますか？ A. おおむね守られますが、保証ではありません。SKIP LOCKED はロック中の行を飛ばすので、先頭の行が別のワーカーに取られていれば次の行が来ます。公式ドキュメントも READ COMMITTED で ORDER BY とロック句を併用したときに順序が崩れて見える場合があると注意しています。厳密な順序が要る処理には向きません。

Q. 取得と処理を1つのトランザクションに入れるべきですか？ A. 処理が短く、結果を同じ DB に書くならその設計が単純です。処理が長い・外部 API を待つ・ワーカーが多い場合は、接続と「idle in transaction」を抱え、autovacuum も進まなくなるので、この記事のように running の状態列で持ち、locked_at で回収する設計が向きます。

Q. ワーカーが落ちたジョブはどう戻りますか？ A. 状態列で持つ設計では自動では戻りません。status = 'running' AND locked_at < now() - interval '60 seconds' の行を queued に戻す UPDATE を定期的に実行します。トランザクションで持つ設計では接続が切れた時点でロールバックされますが、サーバーが切断に気づくまで（client_connection_check_interval が 0 なら次に結果を送るときまで）ロックは残ります。

Q. バッチで何件ずつ取るのがよいですか？ A. 手元では 1 件ずつ 1,430 件/秒、10 件ずつ 10,842 件/秒、50 件ずつ 26,726 件/秒でした。ただし 1 バッチの処理が長いほど「落ちたときに戻すジョブ」が増え、回収の猶予も伸ばす必要があります。1 バッチの処理時間が数秒以内に収まる件数が目安です。

Q. 空のキューをポーリングし続けて問題ありませんか？ A. 空振りの取得クエリは 0.05 ミリ秒前後で、0.5〜1 秒間隔なら数十ワーカーでも負荷は小さいです。着手までの遅れを詰めたいときは INSERT のトリガーで pg_notify を送り、ワーカーは LISTEN で待ちます（手元では通知の遅れは中央値 0.52 ミリ秒）。

Q. テーブルが太ってきました。 A. 1 ジョブで UPDATE 2 回・HOT 更新 0 なので、dead tuple が速く増えます。jobs テーブルだけ autovacuum_vacuum_scale_factor を 0.01 程度に下げ、done の行は別テーブルに移すか定期的に削除します。VACUUM 前は取得クエリが 462 ブロック、後は 4 ブロックでした。

Q. MySQL や他の DB でも同じ書き方ができますか？ A. MySQL 8.0 の InnoDB にも FOR UPDATE SKIP LOCKED と NOWAIT があり、公式リファレンス（Locking Read Concurrency with NOWAIT and SKIP LOCKED）に説明があります。Oracle Database にも SKIP LOCKED があります。この記事の状態遷移と回収の設計はそのまま使えますが、部分インデックスや RETURNING の有無、ロックの実装（InnoDB はインデックスレコードロック）が違うので、EXPLAIN と実測は DB ごとに取り直します。

Q. Redis や SQS ではなく PostgreSQL を選ぶ決め手は？ A. 「ジョブの投入を業務データの更新と同じトランザクションで確定したい」ことです。別のキューに出す場合は、DB 更新と送信の二重書き込みを避けるために Outbox パターンが要ります。処理量は手元で Redis が 1 件ずつで約 5 倍速く、桁違いの件数や多言語のワーカーが要るなら専用キューを選びます。

まとめ

• 取得は UPDATE … WHERE id IN (SELECT id … ORDER BY … LIMIT n FOR UPDATE SKIP LOCKED) RETURNING * の 1 文で行う。ロック無しは 5,000 件中 3,616 件が二重取得、SKIP LOCKED は 1〜8 並列で 0 件だった。
• NOWAIT は 55P03 のエラー、素の FOR UPDATE はコミットまで待ち、待ち明けに別の行を返して待ち時間が無駄になる。子テーブルがあるなら FOR NO KEY UPDATE SKIP LOCKED（FK の INSERT が 2,403 ms → 3.5 ms）。
• 再試行は run_at を指数で後ろにずらすだけ、置き去りの running は locked_at で回収する。処理は冪等に書き、DONE の UPDATE には AND status = 'running' を付ける。
• 処理量は 2 CPU の手元で 1 件ずつ 1,430 件/秒、10 件ずつ 10,842 件/秒。バッチで取る効果が最も大きい。部分インデックスが無いと取得が 0.09 ms → 71 ms。
• 1 ジョブで UPDATE 2 回・HOT 0 で dead tuple が速く増え、VACUUM 前は取得が 462 ブロック・30 倍遅かった。jobs だけ autovacuum を攻め、done を溜めない。

参考・出典

確認日はいずれも 2026年11月4日です。

• PostgreSQL 16 Documentation「SELECT」The Locking Clause：https://www.postgresql.org/docs/16/sql-select.html#SQL-FOR-UPDATE-SHARE
• PostgreSQL 16 Documentation「13.3. Explicit Locking」13.3.2 Row-Level Locks：https://www.postgresql.org/docs/16/explicit-locking.html#LOCKING-ROWS
• PostgreSQL 16 Documentation「25.1. Routine Vacuuming」：https://www.postgresql.org/docs/16/routine-vacuuming.html
• PostgreSQL 16 Documentation「73.7. Heap-Only Tuples (HOT)」：https://www.postgresql.org/docs/16/storage-hot.html
• PostgreSQL 16 Documentation「NOTIFY」：https://www.postgresql.org/docs/16/sql-notify.html
• PostgreSQL 16 Documentation「20.3. Connections and Authentication」client_connection_check_interval：https://www.postgresql.org/docs/16/runtime-config-connection.html
• PostgreSQL 16 Documentation「54.12. pg_locks」：https://www.postgresql.org/docs/16/view-pg-locks.html
• PostgreSQL 16 Documentation「F.33. pgstattuple」：https://www.postgresql.org/docs/16/pgstattuple.html
• psycopg 3「Asynchronous notifications」：https://www.psycopg.org/psycopg3/docs/advanced/async.html#asynchronous-notifications
• Redis「Redis Streams」（コンシューマグループ・XPENDING・XAUTOCLAIM）：https://redis.io/docs/latest/develop/data-types/streams/
• Redis「LMOVE」：https://redis.io/docs/latest/commands/lmove/
• Amazon SQS Developer Guide「Amazon SQS visibility timeout」：https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
• Amazon SQS Developer Guide「Amazon SQS FIFO queues」（スループットの上限）：https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fifo-queues.html
• MySQL 8.0 Reference Manual「Locking Read Concurrency with NOWAIT and SKIP LOCKED」：https://dev.mysql.com/doc/refman/8.0/en/innodb-locking-reads.html

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。