pgmqとは？PostgreSQLをメッセージキューにする拡張の使い方とSupabase Queues連携

pgmq は、すでに動いている PostgreSQL をそのままメッセージキューとして使うための拡張です。Redis や RabbitMQ を別途立てずに、テーブルと SQL 関数だけで AWS SQS 相当のキューを作れます。Supabase の「Queues」機能もこの pgmq を採用しています。基本 SQL・導入方法・2026年に入って追加された FIFO とトピックルーティング・Supabase での有効化手順を、公式リポジトリとドキュメントの記述に沿って整理します。

まとめ

• pgmq は PostgreSQL 拡張。外部ミドルウェアもワーカーも不要で、キュー1本が pgmq.q_<キュー名> というテーブル1枚に対応する。
• 覚える SQL は実質6つ。pgmq.create / pgmq.send / pgmq.read / pgmq.delete / pgmq.archive / pgmq.metrics。
• 対応 PostgreSQL は 14〜18。拡張を入れられないマネージド環境では SQL のみでも導入できる（v1.5.1 以降）。
• 最新は v1.12.0（2026-07-14）。FIFO は v1.9.0、AMQP 風トピックルーティングは v1.11.0 の新機能で、2024年時点の解説記事には載っていない。
• Supabase Queues の実体は pgmq。ただしクライアントから叩ける pgmq_public スキーマは6関数だけで、キューの作成・削除は SQL かダッシュボードから行う。
• 判断基準は「業務データの保存とジョブ投入を同一トランザクションに入れたいか」。DB と独立した可用性やイベントの長期再生が要るなら Kafka や SQS を選ぶ。

pgmqの正体とキューの実体

設計上の要点は、独立したブローカープロセスを一切持たないことです。README は特徴の筆頭に「Lightweight – No background worker or external dependencies, just Postgres SQL objects」を挙げています。pgmq が提供するのは、テーブルと関数とインデックスというありふれた SQL オブジェクトだけです。

キューを1本作ると、pgmq スキーマに q_ を接頭辞にしたテーブルが1枚できます。my_queue なら pgmq.q_my_queue、アーカイブ先は pgmq.a_my_queue です。どちらも普通のテーブルなので、詰まったキューの中身は SELECT でそのまま覗けます。障害調査でブローカーの管理コンソールを開く必要がありません。データ構造としてのキュー（FIFO）とスタック（LIFO）の違いを押さえていれば、挙動の理解はそのまま通用します。なおキュー名は47文字までで、これはパーティション化の有無にかかわらず共通の制限です。

可視性タイムアウト（vt）が二重処理を防ぐ仕組み

pgmq が保証するのは「可視性タイムアウトの範囲内での exactly once 配信」です。メッセージを読むと一定時間そのメッセージが他のコンシューマから見えなくなり、その間に削除もアーカイブもされなければ再び見えるようになります。

vt は処理にかかる想定時間より長く設定します。短すぎると処理中のメッセージが復活して二重実行になり、長すぎるとワーカーが落ちたときの再試行が遅れる。README も「It is recommended to set a vt value that is greater than the expected time it takes to process a message.」と明記しています。処理が終わったら pgmq.delete() で消すか pgmq.archive() で退避させるまでが1サイクルです。

pgmqの導入方法と3つの選択肢

導入方法は2通りあり、選択基準は「Postgres が動いているホストのファイルシステムを触れるか」です。公式の INSTALLATION.md は「Using PGMQ as a Postgres extension is preferred, but the SQL-only approach can also be useful when extension installation is restricted or not available.」として、拡張インストールを推奨しています。

拡張としてインストールする最短経路

試すだけなら、pgmq が同梱された公式 Docker イメージが最短です。この記事で扱う FIFO とトピックは v1.11 系以降の機能なので、タグは最新の v1.12.0 を指定してください（README の例は v1.10.0 のままなので、そのままコピーすると後半の SQL が動きません）。

docker run -d --name pgmq-postgres -e POSTGRES_PASSWORD=postgres -p 5432:5432 ghcr.io/pgmq/pg18-pgmq:v1.12.0

接続して拡張を有効化します。

psql postgres://postgres:postgres@localhost:5432/postgres

CREATE EXTENSION pgmq;

既存の Postgres に入れる場合は pgxn CLI などで拡張ファイルを配置してから CREATE EXTENSION します。入っているバージョンは \dx pgmq または SELECT extversion FROM pg_extension WHERE extname = 'pgmq'; で確認でき、アップグレードは ALTER EXTENSION pgmq UPDATE です。

拡張が使えないマネージド環境ではSQL-onlyで入れる

RDS のようにホストへ触れない環境向けに、v1.5.1（2025-03-21）で「Enable PGMQ to run without installing an extension」が入りました。SQL 定義ファイルを流し込むだけで pgmq スキーマと必要なオブジェクトが作られます。

git clone https://github.com/pgmq/pgmq.git
cd pgmq
psql -f pgmq-extension/sql/pgmq.sql postgres://postgres:postgres@localhost:5432/postgres

ただし公式は、この素の SQL-only 導入について「This installation approach is not versioned and only works for a fresh installation of pgmq.」と警告しています。バージョン追跡もアップグレードもできず、pg_extension にも行が現れないため、どの版が入っているかを後から確認する手段がありません。将来の更新を見込むなら、Rust クライアントの pgmq-cli のようなバージョン付き SQL-only インストールを使ってください。

拡張・SQL-only・バージョン付きSQL-onlyの選択軸

公式の結論は「Use the extension installation when your environment allows it.」です。Supabase は拡張として導入しているため、後述の版確認は pg_extension で行えます。

pgmqの基本操作 ― 送信・受信・削除のSQL

API は AWS SQS と RSMQ に寄せてあり、メッセージは jsonb で扱います。

キュー作成とメッセージ送信

SELECT pgmq.create('my_queue');

-- 即時に読めるメッセージ
SELECT * FROM pgmq.send('my_queue', '{"foo": "bar1"}');

-- 5秒後まで読めないメッセージ
SELECT * FROM pgmq.send(
  queue_name => 'my_queue',
  msg        => '{"foo": "bar2"}',
  delay      => 5
);

戻り値は採番された msg_id です。delay は秒数のほか timestamptz も受け付けるので、CURRENT_TIMESTAMP + INTERVAL '1 day' と書けば「明日から読めるメッセージ」になります。複数件をまとめて入れるなら pgmq.send_batch() に jsonb 配列を渡します。トレース ID などのメタデータは本文と分けて headers 引数に載せられます。

read・pop・read_with_pollの使い分け

-- 2件読み、30秒間は他のコンシューマから隠す
SELECT * FROM pgmq.read(
  queue_name => 'my_queue',
  vt         => 30,
  qty        => 2
);

返るのは msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers の7列です。キューが空、もしくは全メッセージが不可視なら0行が返ります。

読んだ瞬間に消してよい軽い処理なら pgmq.pop('my_queue') の1行で済みます。ただし pop は取り出しと同時に削除するため、処理中にワーカーが落ちるとメッセージは失われる。再試行を効かせたいなら必ず read と delete のペアを使ってください。

空振りのポーリングを減らしたい場合は pgmq.read_with_poll() です。メッセージが無いときは max_poll_seconds（既定5秒）だけ待ち、その間に到着すれば即座に返します（内部間隔は poll_interval_ms、既定100ミリ秒）。なお read 系の conditional 引数は公式が「This feature is experimental, and the API is subject to change in future releases」と明記しているため、本番の分岐ロジックを預けるべきではありません。

deleteとarchiveの選択、metricsでの監視

処理完了時の後始末は2択です。pgmq.delete('my_queue', 6) は完全に消し、pgmq.archive() はキューから外して pgmq.a_my_queue へ移します。監査ログやリプレイの要件があるならアーカイブ一択です。pgmq.archive(queue_name => 'my_queue', msg_ids => ARRAY[3, 4, 5]) のように複数件をまとめて処理できます。キューごと消すなら pgmq.drop_queue('my_queue')、中身だけ空にするなら pgmq.purge_queue() です。

詰まりの検知には pgmq.metrics('my_queue') を使います。返るのは滞留件数（queue_length）、今すぐ読める件数（queue_visible_length）、最古・最新メッセージの経過秒数、通算処理件数です。監視で見るべきは oldest_msg_age_sec の伸びで、ここが増え続けていればコンシューマがワークロードに追いついていません。全キューをまとめて見るなら pgmq.metrics_all() です。

v1.9.0以降で追加された機能 ― FIFO・トピック・INSERT通知

2024年に書かれた pgmq の記事は、この表の1行目より前の世代を説明しています。手元の環境でどこまで使えるかは、拡張として導入していれば SELECT extversion FROM pg_extension WHERE extname = 'pgmq'; で判別できます。素の SQL-only 導入では版が記録されないため、この確認自体ができません。

x-pgmq-groupヘッダによる順序保証

FIFO は v1.9.0 で入りました。キューの種類を変えるのではなく、送信時のヘッダでグループを指定する方式なので、既存キューはそのまま動きます。

SELECT pgmq.send('my_queue', '{"order": 1}', headers => '{"x-pgmq-group": "user123"}');
SELECT pgmq.send('my_queue', '{"order": 2}', headers => '{"x-pgmq-group": "user123"}');

同じグループ ID を持つメッセージは送信順に処理され、グループが違えば並列に処理できます。読み出しは目的別に用意されており、read_grouped_rr() はグループ間をラウンドロビンで公平に混ぜるためマルチテナント向き、read_grouped() は最古のグループからバッチを埋めるスループット重視です。各グループの先頭を1件ずつ返す read_grouped_head() は v1.11.1、そのポーリング版は v1.12.0 の追加なので、v1.9 系や v1.10 系の環境には存在しません。ヘッダを付けないメッセージは単一のデフォルトグループとして扱われます。

トピックルーティングで1メッセージを複数キューへ配る

v1.11.0 で AMQP のトピックエクスチェンジに似たルーティングが入りました。ドット区切りのルーティングキーとパターンをバインドし、マッチしたすべてのキューへ同じメッセージを配ります。

SELECT pgmq.bind_topic('logs.#', 'all_logs');
SELECT pgmq.bind_topic('logs.*.error', 'error_logs');

-- all_logs と error_logs の両方へ配送される
SELECT pgmq.send_topic('logs.db.error', '{"message": "DB connection failed"}');

ワイルドカードは2種類で、* はちょうど1セグメント、# は0個以上のセグメントにマッチします。logs.* は logs.error にマッチしますが logs.api.error にはマッチしません。パターンはバインド時に正規表現へプリコンパイルされ、キューを drop すればバインドも CASCADE で消えます。pgmq.test_routing() を使えば、本番キューへ1件も送らずに配送先だけを列挙できます。この pub/sub 的な使い方が必要になった時点で、RabbitMQ と Kafka の設計思想の違いと照らし合わせて選定を見直す価値があります。

ポーリングをやめてLISTENで待つ

pgmq.enable_notify_insert('my_queue') を実行すると、メッセージ挿入時に pgmq.q_my_queue.INSERT チャネルへ NOTIFY が飛びます。コンシューマ側は LISTEN "pgmq.q_my_queue.INSERT"; で待機し、通知が来たときだけ読みにいけます。

効くのは低頻度・散発的なキューです。公式も「Notifications are most valuable for queues with sporadic or low-volume traffic.」と用途を限定しています。高頻度時は通知が洪水になるため、第2引数のスロットル間隔（既定250ミリ秒）でまとめられます。0 を渡すとスロットリングは無効。常時メッセージが流れているキューなら、通知を仕込まず素直にポーリングし続けたほうが構成はシンプルです。

Supabase Queuesでのpgmq利用手順

Supabase Queues は 2024年12月5日に発表された機能で、公式は「a Postgres-native durable Message Queue system with guaranteed delivery built on the pgmq database extension」と説明しています。実体は pgmq なので、上で見た SQL はそのまま使えます。Supabase 自体の位置づけはSupabase と Firebase の比較で整理しています。

Queuesモジュールの有効化とキュー作成

ダッシュボードの Integrations にある Queues Postgres Module から pgmq 拡張を有効化します。公式 Quickstart は Postgres 15.6.1.143 以降が必要と明記しているので、古いプロジェクトはまず DB バージョンを確認してください。

Queues ページの Create queue でキュー名を決め、種類を選びます。Quickstart は「three types of Queues」と書いていますが、実際に列挙されているのは Basic Queue と Unlogged Queue の2種類だけです。Basic は logged テーブルの耐久性重視、Unlogged は unlogged テーブルの性能重視で、pgmq の pgmq.create() と pgmq.create_unlogged() に対応します。見落としやすいのは、Unlogged を選んでもアーカイブ側は守られる点です。Quickstart は「The pgmq.a_<queue_name> table is still created as a logged table」と明記しています。作成時の Row Level Security は有効のままに。有効なら、キューテーブルへ追加のポリシーを書く必要はありません。

pgmq_publicスキーマの公開とロール権限

Queues の Settings で「Expose Queues via PostgREST」を有効にすると pgmq_public スキーマが作られ、ここに置かれたラッパー関数だけが Data API 経由で呼べるようになります。公開されるのは send、send_batch、read、pop、archive、delete の6つです。公式ドキュメントは「this schema exposes operations to manage Queue Messages to consumers client-side, but does not expose functions for creating or dropping Queues.」と明言しており、キューの作成・削除はクライアントからは行えません。Data API へ公開する場合に限り、キューの概要ページの Add RLS Policy からクライアントに触らせるキューのポリシーを追加します。

authenticated ロールに与える権限は操作ごとに異なります。send と send_batch は Select と Insert、read と pop は Select と Update、archive と delete は Select と Delete です。read が Update を要求するのは、読み出し時に可視性タイムアウトを書き戻すためです。ここを Select だけにすると、送信は通るのに受信だけが権限エラーで落ちます。

supabase-jsからの呼び出し

import { createClient } from '@supabase/supabase-js'

const supabase = createClient('supabaseURL', 'supabaseKey')

const result = await supabase.schema('pgmq_public').rpc('send', {
  queue_name: 'foo',
  message: { hello: 'world' },
  sleep_seconds: 30,
})

.schema('pgmq_public') を挟まないと、PostgREST は既定の public スキーマを探しにいって関数が見つかりません。受信は rpc('pop', { queue_name: 'foo' }) になります。セルフホストやローカル開発の場合はダッシュボードの設定が無いため、Supabase CLI なら config.toml の schemas に、Docker Compose なら環境変数 PGRST_DB_SCHEMAS に pgmq_public を追加してください。Inngest と Supabase を組み合わせたバックグラウンド処理のように外部のジョブ基盤を使う構成と比べると、pgmq はキューだけを DB 内に閉じ込められる点が違いになります。

他のキュー基盤との使い分け

この表の2行目が pgmq を選ぶ最大の理由です。業務データの INSERT とジョブの enqueue を同一トランザクションに入れられるので、「データは保存されたのにジョブが投入されていない」という不整合が構造的に起きません。外部ブローカーではアウトボックスパターンを自前で組む必要がある部分です。

pgmqを避けたほうがよい3つの条件

逆に、次のいずれかに当てはまるなら pgmq は選ばないでください。第一に、キューの書き込みが本業のクエリと Postgres の I/O を食い合う規模の場合。キューは高頻度の INSERT と UPDATE と DELETE を繰り返すため、VACUUM の負荷がそのまま本番 DB に乗ります。第二に、DB がダウンしてもキューだけは受け付け続けたい要件がある場合。pgmq に障害分離は存在しません。第三に、長時間実行のワークフローを状態遷移込みで管理したい場合です。可視性タイムアウトによる再試行は短い処理の失敗をやり直す仕組みであり、数時間かかる多段処理には向きません。この用途にはTemporal のようなワークフローエンジンが構造的に合致します。

pgmq運用の落とし穴

アーカイブテーブルの肥大化

pgmq.archive() はメッセージを消さずに a_ テーブルへ積み続けるので、放置すると容量を圧迫します。保持期間を決めて定期削除するか、pgmq.create_partitioned() でパーティション化して古いパーティションごと落とす設計にしてください。パーティション化には pg_partman 拡張が別途必要です。なお pgmq.detach_archive() は現在 no-op で、v2.0 での削除が予告されています。

read_ctで毒メッセージを検知する

read_ct はメッセージが読まれた回数を持つ列です。値が異常に大きいメッセージは、毎回失敗して vt 切れで復活し続けている毒メッセージです。pgmq にデッドレターキューの組み込み機能は無いため、read_ct が閾値を超えたら別キューへ退避させる処理を自分で書く必要があります。

マネージド環境の版固定

マネージドサービス上の pgmq は提供側が固定したバージョンで動くため、API リファレンスに載っている FIFO やトピックが使えるとは限りません。ドキュメントを読む前に pg_extension で実際の版を確認する順序にしてください。PostgreSQL の拡張である pgvector など、他の拡張でも同じ落とし穴があります。

よくある質問

pgmqとSupabase Queuesの違いは何ですか

pgmq が本体のオープンソース拡張で、Supabase Queues はそれをホスティング環境に統合した機能名です。キューの実装そのものは同一で、Supabase 側の追加分はダッシュボードの管理画面、Row Level Security との統合、pgmq_public による Data API 公開の3点です。

pgmqの公式ドキュメントとSQL関数の一覧はどこにありますか

ドキュメントは pgmq.github.io/pgmq、ソースは github.com/pgmq/pgmq です。旧 URL の tembo-io/pgmq は現在のリポジトリへリダイレクトされます。関数一覧は docs 配下の API リファレンスにあり、送信系・読み出し系・削除系・キュー管理・ユーティリティに分類されています。

デッドレターキューはどう実装しますか

組み込み機能が無いため自作します。read で取得した read_ct が閾値（3〜5回程度）を超えたメッセージを、専用の失敗キューへ pgmq.send() し直すか、pgmq.archive() でアーカイブ表へ落として元キューから外す実装が基本形です。アーカイブ表は普通のテーブルなので、失敗メッセージの調査はそのまま SQL で行えます。

Supabase Queuesに追加料金はかかりますか

公式ブログは「Supabase Queues runs entirely in your database so there’s no additional costs to use the functionality.」としています。キュー機能そのものへの課金は無く、消費するのはプロジェクトのデータベース容量と計算リソースです。

関連記事

• RabbitMQとKafkaの違いとは？設計思想・性能・使い分けの判断基準を実装視点で解説
• スタックとキューの違い｜FIFO・LIFOと使い分け・Python実装を図解
• Inngestとは？使い方・インストール（npm/CLI）とNext.js・Vercel・Supabase連携を解説
• Temporal Workflowとは？コードで書く分散ワークフローエンジンの仕組みと比較
• SupabaseとFirebaseの違いを徹底比較｜料金・認証・どっちを選ぶか【2026年版】