--- title: "pgmqとは？PostgreSQLをメッセージキューにする拡張の使い方とSupabase Queues連携" url: "https://www.issoh.co.jp/tech/details/4240/" published: 2024-11-18 updated: 2026-07-28 categories: ["データベース"] publisher: "株式会社一創" --- # 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）の違い](/tech/details/4587/)を押さえていれば、挙動の理解はそのまま通用します。なおキュー名は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の選択軸 | 比較軸 | 拡張インストール | SQL-only（素） | SQL-only（バージョン付き） | | --------- | -------- | ----------- | ----------------- | | バージョン追跡 | 可 | 不可 | 可 | | アップグレード | 可 | 不可 | 可 | | ホストのFS操作 | 必要 | 不要 | 不要 | | マネージドDB対応 | 限定的 | 対応 | 対応 | | 向く環境 | 自前運用 | 制約のある環境 | 制約のある環境 | 公式の結論は「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通知 | 版 | リリース日 | 主な追加 | | ------- | ---------- | ------------------------------------------ | | v1.5.1 | 2025-03-21 | 拡張なしのSQL-only導入 | | v1.7.0 | 2025-09-11 | pop複数件、NOTIFY通知 | | v1.8.0 | 2025-11-30 | 通知スロットリング、set\_vtバッチ | | v1.9.0 | 2026-01-18 | FIFOキュー（read\_grouped / read\_grouped\_rr） | | v1.10.0 | 2026-02-02 | last\_read\_at列、set\_vtのtimestamptz対応 | | v1.11.0 | 2026-02-19 | トピックルーティング | | v1.11.1 | 2026-04-19 | read\_grouped\_head | | v1.12.0 | 2026-07-14 | read\_grouped\_head\_with\_poll | 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 の設計思想の違い](/tech/details/15305/)と照らし合わせて選定を見直す価値があります。 ### ポーリングをやめて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 の比較](/column/details/3002/)で整理しています。 ### 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_` 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 を組み合わせたバックグラウンド処理](/tech/details/2595/)のように外部のジョブ基盤を使う構成と比べると、pgmq はキューだけを DB 内に閉じ込められる点が違いになります。 ## 他のキュー基盤との使い分け | 観点 | pgmq | RabbitMQ | Kafka | Amazon SQS | | ------------- | ------ | -------- | ----- | ---------- | | 追加インフラ | 不要 | 必要 | 必要 | 不要（従量課金） | | DBトランザクションと同一 | 可 | 不可 | 不可 | 不可 | | メッセージの直接SQL参照 | 可 | 不可 | 不可 | 不可 | | DB障害時の独立性 | 無し | 有り | 有り | 有り | | 再生・長期保持 | アーカイブ表 | 限定的 | 得意 | 限定的 | この表の2行目が pgmq を選ぶ最大の理由です。業務データの INSERT とジョブの enqueue を同一トランザクションに入れられるので、「データは保存されたのにジョブが投入されていない」という不整合が構造的に起きません。外部ブローカーではアウトボックスパターンを自前で組む必要がある部分です。 ### pgmqを避けたほうがよい3つの条件 逆に、次のいずれかに当てはまるなら pgmq は選ばないでください。第一に、キューの書き込みが本業のクエリと Postgres の I/O を食い合う規模の場合。キューは高頻度の INSERT と UPDATE と DELETE を繰り返すため、VACUUM の負荷がそのまま本番 DB に乗ります。第二に、DB がダウンしてもキューだけは受け付け続けたい要件がある場合。pgmq に障害分離は存在しません。第三に、長時間実行のワークフローを状態遷移込みで管理したい場合です。可視性タイムアウトによる再試行は短い処理の失敗をやり直す仕組みであり、数時間かかる多段処理には向きません。この用途には[Temporal のようなワークフローエンジン](/tech/details/9090/)が構造的に合致します。 ## 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](/tech/details/6231/) など、他の拡張でも同じ落とし穴があります。 ## よくある質問 ### 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の違いとは？設計思想・性能・使い分けの判断基準を実装視点で解説](/tech/details/15305/) - [スタックとキューの違い｜FIFO・LIFOと使い分け・Python実装を図解](/tech/details/4587/) - [Inngestとは？使い方・インストール（npm/CLI）とNext.js・Vercel・Supabase連携を解説](/tech/details/2595/) - [Temporal Workflowとは？コードで書く分散ワークフローエンジンの仕組みと比較](/tech/details/9090/) - [SupabaseとFirebaseの違いを徹底比較｜料金・認証・どっちを選ぶか【2026年版】](/column/details/3002/) --- 出典: [pgmqとは？PostgreSQLをメッセージキューにする拡張の使い方とSupabase Queues連携]()（株式会社一創）