This chapter explains how a message queue enables asynchronous communication between producers and consumers and shows when Postgres can serve as a capable queue. Because messages can be modeled as rows, persisted durably, and manipulated with SQL and stored procedures, Postgres provides the essential building blocks. The chapter advises using Postgres as a queue when you need transactional atomicity between business data and emitted messages, when expected traffic fits within the database’s read/write capacity (with replicas for reads and scale-up or distributed options for heavy writes), and when you already run Postgres and prefer to avoid introducing a separate messaging system.
The chapter then builds a simple, durable queue on a table with id, message payload, timestamps, and status, plus three functions: mq.enqueue to add messages, mq.dequeue to retrieve them in FIFO order while using row locks and SKIP LOCKED for concurrency, and mq.mark_completed to finalize or delete processed items. A DMV scenario illustrates end-to-end flow: terminals enqueue visitors, officers dequeue and work them, and statuses transition from new to processing to completed. To notify workers of new arrivals without polling, the chapter integrates LISTEN/NOTIFY by emitting a notification from mq.enqueue, outlining practical nuances of client behavior and primary-replica constraints.
Finally, it covers operational considerations for higher throughput: indexing (for example, on created_at and status) to accelerate dequeues, partitioning to keep large tables performant and ease retention, and failover patterns that prevent messages from getting stuck in processing. The chapter concludes with pgmq, a lightweight extension that offers an SQS-like API and production features out of the box: per-queue tables, jsonb payloads, headers, read counters, and visibility timeouts for automatic reappearance of unacknowledged messages, plus functions such as send, read, pop, and archive. Using the DMV example, it demonstrates visibility timeouts to requeue no-shows and archival for completed work, showing that you can either roll your own with core Postgres or adopt pgmq for a ready-made solution.
Summary
Postgres can be used as a message queue to exchange messages, events, and tasks between producers and consumers.
A custom message queue can be implemented on top of a regular Postgres table, allowing you to tailor the implementation to the specifics of your messaging scenario.
The custom implementation can rely on FOR UPDATE SKIP LOCKED to support concurrent access by multiple consumers.
The LISTEN and NOTIFY statements can be used to notify consumers when new messages are added to the queue.
There are several considerations when using Postgres as a message queue, including partitioning, indexing, and failover handling for message processing.
The pgmq extension provides a ready-to-use message queue for Postgres with API parity with AWS SQS (Simple Queue Service).
FAQ
When should I use Postgres as a message queue instead of a dedicated system?Use Postgres when all or most of the following apply:
- You need transactional atomicity between a business operation and the message it emits (one commit guarantees both happen or neither).
- The expected message volume and concurrency can be handled by your Postgres setup (single primary for writes; replicas can offload reads; scale up or consider distributed Postgres if writes saturate).
- You already run Postgres and prefer not to introduce and operate another technology.
If the required optimization and scaling effort becomes high or complex, consider a purpose-built queue.How do I model a simple FIFO queue table in Postgres?Create a regular table with:
- id: BIGSERIAL (or UUID for high-concurrency ID generation) as primary key
- message: JSON payload from producers
- created_at: timestamptz with default NOW(), used for FIFO ordering
- status: enum with values new, processing, completed
This schema gives durable storage, clear ordering, and lifecycle tracking while staying minimal.What functions do producers and consumers call in the custom design?- mq.enqueue(new_message JSON): inserts a message (status = new). Can be called inside a broader transaction to ensure atomicity with business logic.
- mq.dequeue(messages_cnt INT): returns up to messages_cnt oldest new messages, marks them processing, and hides them from other consumers.
- mq.mark_completed(message_ids BIGINT[], to_delete BOOLEAN = FALSE): marks processed messages as completed or deletes them immediately when to_delete = TRUE.How does mq.dequeue prevent multiple consumers from processing the same message?It uses SELECT ... FOR UPDATE SKIP LOCKED to:
- Lock candidate rows so only one transaction can update them at a time.
- Skip rows already locked by other consumers.
Then it updates status to processing and returns those rows. Locks are released on transaction commit/rollback (explicit or implicit), ensuring safe parallel consumption without blocking.Why store the payload as JSON instead of JSONB?For a pass-through queue, JSON is often better because:
- It avoids JSONB preprocessing overhead, speeding ingestion.
- It preserves the original structure (e.g., key order) exactly as sent by producers.
Use JSONB when you need to index, traverse, or transform the payload inside Postgres.How can I notify workers about new messages using Postgres?Use LISTEN/NOTIFY:
- Workers LISTEN on a channel (e.g., queue_new_message).
- Producers (or triggers/functions) NOTIFY (or pg_notify) on that channel after enqueue.
Important limitations:
- No backlog: only currently listening sessions receive notifications.
- Not supported on replicas: subscribe on the primary (keep a dedicated connection if you use read replicas).
Note: psql shows notifications when the next command runs; application drivers typically deliver them asynchronously.What indexing helps mq.dequeue scale?Add secondary indexes to avoid full scans:
- A btree index on created_at to fetch the earliest message efficiently.
- Or a partial index on (created_at, status) WHERE status = 'new' to narrow to dequeuable rows.
The primary key on id already helps mq.mark_completed locate rows quickly.How do I handle messages stuck in the processing state (consumer failover)?Add a recovery mechanism:
- Periodically detect messages that have been processing too long and reset their status to new (e.g., with pg_cron running a simple UPDATE based on a threshold).
- Alternatively, use a visibility timeout (vt) approach like pgmq.read provides, where messages automatically reappear after a timeout if not completed.When and how should I partition a high-volume queue?Partition when table size and churn (inserts/updates/deletes) start to affect performance or autovacuum lag:
- Range partition by created_at to keep hot data small and drop old partitions quickly.
- Optionally partition by status or by (created_at, status).
- Tools like pg_partman (for auto-partition management) and pg_cron (for scheduling) can automate creation and cleanup.
Partitioning reduces planning/scan overhead and mitigates dead-tuple buildup.What does the pgmq extension provide, and how do I use it?pgmq is a lightweight, SQS-like queue for Postgres that abstracts table design and best practices:
- Create a queue: pgmq.create('queue_name') → underlying table q_queue_name (plus archive table a_queue_name).
- Send a message: pgmq.send(queue_name, msg [and optional headers/delay]).
- Receive without deleting: pgmq.read(queue_name, vt, qty, conditional) sets a visibility timeout (vt) so messages reappear if not completed.
- Receive and delete: pgmq.pop(queue_name).
- Complete by archiving: pgmq.archive(queue_name, msg_id) moves the message to the archive table.
It also adds helpful columns (read_ct, vt, headers), default indexes, and functions for filtering and retention-like workflows.
pro $24.99 per month
access to all Manning books, MEAPs, liveVideos, liveProjects, and audiobooks!
Just Use Postgres! ebook for free
