{{tag>[database postgresql extension queue pgmq sql]}}

====== PGMQ ======
PGMQ(Postgres Message Queue)는 PostgreSQL 내부의 SQL object만으로 message queue를 제공하는 extension이다. 별도 background worker 없이 queue 생성, JSON message 송수신, visibility timeout, archive, FIFO group, metrics를 SQL function으로 다룬다.

===== Summary =====
  * message는 queue별 ''pgmq.q_QUEUE_NAME'' table에 저장되며 명시적으로 delete 또는 archive할 때까지 유지된다.
  * consumer가 message를 읽으면 visibility timeout 동안 다른 consumer에게 보이지 않는다.
  * 처리 성공 후 ''pgmq.delete()'' 또는 ''pgmq.archive()''를 호출해야 한다.
  * 현재 공식 지원 범위는 PostgreSQL 14–18이다.

===== Installation =====

==== Docker ====
공식 image에는 PGMQ extension이 미리 설치되어 있다. 아래 password는 local test용 예시이며 production에서는 안전한 secret 전달 방식을 사용한다.

<code bash>
docker run -d \
  --name pgmq-postgres \
  -e POSTGRES_PASSWORD=postgres \
  -p 5432:5432 \
  ghcr.io/pgmq/pg18-pgmq:v1.10.0

psql 'postgres://postgres:postgres@localhost:5432/postgres'
</code>

<code postgresql>
CREATE EXTENSION pgmq;
</code>

==== PGXN ====
PostgreSQL server host에서 ''pg_config''가 ''PATH''에 잡힌 상태로 설치한다.

<code bash>
pgxn install pgmq
</code>

==== Source build ====
<code bash>
git clone https://github.com/pgmq/pgmq.git
cd pgmq/pgmq-extension
make
sudo make install
</code>

설치 후 사용할 database마다 extension을 활성화한다.

<code postgresql>
SELECT name, default_version, installed_version
FROM pg_available_extensions
WHERE name = 'pgmq';

CREATE EXTENSION pgmq;
</code>

==== SQL-only ====
server filesystem에 extension file을 설치할 수 없는 환경에서는 SQL object만 설치할 수 있다. 아래 unversioned 방식은 fresh installation 전용이므로 upgrade가 필요한 운영 환경에서는 extension 또는 versioned client 방식을 우선 검토한다.

<code bash>
git clone https://github.com/pgmq/pgmq.git
cd pgmq
psql -f pgmq-extension/sql/pgmq.sql 'postgres://USER:PASSWORD@HOST:5432/DBNAME'
</code>

<note important>
공식 upstream은 Debian/Ubuntu APT, RHEL/Fedora DNF/YUM, Homebrew, Windows ''winget'' 전용 PGMQ package를 안내하지 않는다. 해당 platform에서도 PostgreSQL server와 build dependency를 준비한 뒤 PGXN/source 방식을 쓰거나 공식 Docker image를 사용한다.
</note>

==== Verification ====
<code postgresql>
\dx pgmq
SELECT extversion FROM pg_extension WHERE extname = 'pgmq';
SELECT * FROM pgmq.list_queues();
</code>

===== Usage =====
<code postgresql>
SELECT pgmq.create('jobs');
SELECT * FROM pgmq.send('jobs', '{"task":"resize","image_id":42}'::jsonb);
SELECT * FROM pgmq.read('jobs', 30, 1);
SELECT pgmq.archive('jobs', 1);
</code>

  * ''<color #c3c3c3>**pgmq**</color>.<color #ff7f27>create</color>(<color #7092be>QUEUE_NAME</color>)'' queue 생성
  * ''<color #c3c3c3>**pgmq**</color>.<color #ff7f27>send</color>(<color #7092be>QUEUE_NAME</color>, <color #7092be>MESSAGE</color>)'' JSON message 전송
  * ''<color #c3c3c3>**pgmq**</color>.<color #ff7f27>read</color>(<color #7092be>QUEUE_NAME</color>, <color #7092be>VT_SECONDS</color>, <color #7092be>QTY</color>)'' message 읽기
  * ''<color #c3c3c3>**pgmq**</color>.<color #ff7f27>archive</color>(<color #7092be>QUEUE_NAME</color>, <color #7092be>MSG_ID</color>)'' 처리한 message 보관

===== Commands =====

==== Queue lifecycle ====
^ Function ^ Purpose ^ Notes ^
| ''pgmq.create(QUEUE_NAME)'' | 일반 queue 생성 | queue 이름은 최대 47자 |
| ''pgmq.create_non_partitioned(QUEUE_NAME)'' | non-partitioned queue 생성 | ''create()''의 명시적 형태 |
| ''pgmq.create_partitioned(...)'' | partitioned queue 생성 | ''pg_partman'' 필요 |
| ''pgmq.create_unlogged(QUEUE_NAME)'' | unlogged queue 생성 | 성능 우선, crash durability 없음 |
| ''pgmq.list_queues()'' | queue 목록 조회 | partitioned/unlogged 여부 포함 |
| ''pgmq.purge_queue(QUEUE_NAME)'' | queue의 모든 message 영구 삭제 | 삭제 건수 반환 |
| ''pgmq.drop_queue(QUEUE_NAME)'' | queue와 archive table 삭제 | destructive |

<code postgresql>
SELECT pgmq.create('jobs');
SELECT pgmq.create_partitioned('events', '100000', '10000000');
SELECT * FROM pgmq.list_queues();
</code>

<note warning>
''pgmq.purge_queue()''는 queue 내용 전체를, ''pgmq.drop_queue()''는 queue와 archive table을 삭제한다. production에서는 대상 이름과 backup/retention 정책을 확인한 뒤 실행한다.
</note>

==== Send ====
''message''와 선택적 ''headers''는 ''jsonb''다. ''delay''에는 seconds 정수 또는 ''timestamptz''를 전달할 수 있다.

<code postgresql>
-- 단일 message
SELECT * FROM pgmq.send(
  queue_name => 'jobs',
  msg        => '{"task":"resize","image_id":42}'::jsonb
);

-- header와 10초 delay
SELECT * FROM pgmq.send(
  queue_name => 'jobs',
  msg        => '{"task":"email","user_id":7}'::jsonb,
  headers    => '{"trace_id":"abc123"}'::jsonb,
  delay      => 10
);

-- batch
SELECT * FROM pgmq.send_batch(
  'jobs',
  ARRAY['{"task":"one"}', '{"task":"two"}']::jsonb[]
);
</code>

==== Read and acknowledge ====
<code postgresql>
-- 최대 5개를 읽고 60초간 다른 consumer에게 숨긴다.
SELECT * FROM pgmq.read('jobs', 60, 5);

-- queue가 비었으면 최대 5초간 100ms 간격으로 poll한다.
SELECT * FROM pgmq.read_with_poll('jobs', 60, 5, 5, 100);

-- 성공 처리 후 영구 삭제
SELECT pgmq.delete('jobs', 42);

-- 또는 archive table로 이동
SELECT pgmq.archive('jobs', 42);

-- batch acknowledge
SELECT * FROM pgmq.delete('jobs', ARRAY[43, 44]::bigint[]);
SELECT * FROM pgmq.archive('jobs', ARRAY[45, 46]::bigint[]);
</code>

<note important>
''vt''는 예상 처리 시간보다 길게 잡는다. ''vt'' 안에서는 동일 message를 한 consumer에게 전달하지만, 시간이 끝나기 전에 delete/archive하지 않으면 다시 visible 상태가 되므로 consumer 작업은 idempotent하게 설계하는 편이 안전하다.
</note>

''pgmq.pop()''은 읽는 즉시 삭제하므로 consumer가 이후 처리에 실패하면 message를 복구할 수 없다.

<code postgresql>
SELECT * FROM pgmq.pop('jobs');
</code>

==== Visibility timeout ====
처리 시간이 늘어난 message는 현재 시점부터 새 ''vt'' seconds만큼 visibility timeout을 연장할 수 있다.

<code postgresql>
SELECT * FROM pgmq.set_vt('jobs', 42, 120);
SELECT * FROM pgmq.set_vt('jobs', ARRAY[43, 44]::bigint[], 120);
</code>

==== FIFO groups ====
''x-pgmq-group'' header가 같은 message는 group 안에서 순서대로 처리한다. FIFO 조회를 자주 쓰는 queue에는 header GIN index를 만든다.

<code postgresql>
SELECT * FROM pgmq.send(
  'jobs',
  '{"sequence":1}'::jsonb,
  '{"x-pgmq-group":"customer-42"}'::jsonb
);

SELECT pgmq.create_fifo_index('jobs');
SELECT * FROM pgmq.read_grouped('jobs', 60, 10);
SELECT * FROM pgmq.read_grouped_rr('jobs', 60, 10);
</code>

  * ''read_grouped()''는 가장 오래된 available group을 우선 채운다.
  * ''read_grouped_rr()''는 여러 group을 round-robin 방식으로 읽는다.
  * 각 function에는 long-poll variant인 ''read_grouped_with_poll()'', ''read_grouped_rr_with_poll()''이 있다.

==== Metrics ====
<code postgresql>
SELECT * FROM pgmq.metrics('jobs');
SELECT * FROM pgmq.metrics_all();
</code>

주요 field는 ''queue_length'', ''queue_visible_length'', ''newest_msg_age_sec'', ''oldest_msg_age_sec'', ''total_messages'', ''scrape_time''이다.

==== LISTEN / NOTIFY ====
sporadic traffic에서는 insert notification을 활성화해 불필요한 polling을 줄일 수 있다. channel 이름은 ''pgmq.q_QUEUE_NAME.INSERT'' 형태다.

<code postgresql>
SELECT pgmq.enable_notify_insert('jobs', 250);
LISTEN "pgmq.q_jobs.INSERT";

-- 더 이상 notification이 필요하지 않을 때
SELECT pgmq.disable_notify_insert('jobs');
</code>

===== Examples =====

==== Safe consumer transaction ====
아래는 한 message를 읽고 application 처리 결과가 성공했을 때 archive하는 기본 흐름이다. 외부 API 호출처럼 database transaction으로 되돌릴 수 없는 작업에는 별도의 idempotency key와 retry 정책이 필요하다.

<code postgresql>
BEGIN;

SELECT *
FROM pgmq.read('jobs', 60, 1);

-- application이 반환된 msg_id=42의 작업을 성공적으로 처리했다고 가정
SELECT pgmq.archive('jobs', 42);

COMMIT;
</code>

==== Inspect archive ====
archive table은 ''pgmq.a_QUEUE_NAME'' 형식이다.

<code postgresql>
SELECT msg_id, read_ct, enqueued_at, archived_at, message, headers
FROM pgmq.a_jobs
ORDER BY archived_at DESC
LIMIT 20;
</code>

===== Help =====
PGMQ는 독립 CLI가 아니므로 ''pgmq --help''가 없다. 설치된 database에서 실제 function signature와 extension version을 조회한다.

++++ PGMQ function inspection |
<code postgresql>
\dx pgmq
\df pgmq.*
\df+ pgmq.read
\df+ pgmq.send

SELECT p.oid::regprocedure AS signature
FROM pg_proc AS p
JOIN pg_namespace AS n ON n.oid = p.pronamespace
WHERE n.nspname = 'pgmq'
ORDER BY 1;
</code>
++++

===== Troubleshooting =====
  * ''function pgmq... does not exist'': 현재 database에 ''CREATE EXTENSION pgmq''를 실행했는지, ''\dx pgmq''와 ''\df pgmq.*''로 확인한다.
  * message가 다시 보임: consumer가 ''vt'' 안에 delete/archive하지 못한 것이다. 처리 시간과 retry 정책에 맞춰 ''vt''를 늘리거나 ''set_vt()''로 연장한다.
  * queue가 비어 보임: message의 delay 또는 ''vt''가 미래인지 확인하고 ''metrics()''의 visible length를 함께 본다.
  * partitioned queue 생성 실패: compatible ''pg_partman'' extension 설치 및 활성화 여부를 확인한다.
  * FIFO query가 느림: queue에 ''pgmq.create_fifo_index()''를 적용했는지 확인한다.

===== Compatibility =====
  * PGMQ main documentation 기준 PostgreSQL 14–18을 지원한다.
  * partitioned queue는 ''pg_partman''에 의존한다.
  * SQL-only unversioned 설치는 fresh installation 용도이며 extension upgrade path와 동일하게 취급하지 않는다.
  * function overload와 return field는 PGMQ version에 따라 달라질 수 있으므로 배포된 version의 ''\df+ pgmq.*'' 결과를 우선한다.

===== Deprecated / Legacy =====
  * <del>''pgmq.detach_archive(QUEUE_NAME)''</del>는 deprecated no-op이며 PGMQ 2.0에서 제거될 예정이다.
  * <del>''pgmq.drop_queue(QUEUE_NAME, PARTITIONED)''</del> 2-argument form은 deprecated다. partition 여부를 자동 판별하는 single-argument form을 사용한다.

===== See Also =====
  * [[https://pgmq.github.io/pgmq/|PGMQ Documentation]]
  * [[https://pgmq.github.io/pgmq/api/sql/functions/|PGMQ SQL Functions]]
  * [[https://github.com/pgmq/pgmq|PGMQ GitHub Repository]]
  * [[postgresql:ko]]
  * [[postgresql:psql]]
  * [[postgresql:command]]

===== History =====
  * codex:: 2026-08-05 Added PGMQ installation, queue lifecycle, message processing, FIFO, metrics, notification, and operational safety reference.

{{indexmenu>.#1|js}}