A queue delivers messages in FIFO order: consumers receive older messages before newer ones. A PostgreSQL implementation stores each message with an insertion timestamp and lifecycle status. section 11.2
A practical message queue separates producer, consumer, and completion work:
- Enqueue stores the original JSON message and automatically assigns its ID, timestamp, and new status. section 11.2
- Dequeue selects new messages by created_at, locks them with SKIP LOCKED, and changes their status to processing. section 11.2
- Completion either marks selected messages completed or deletes them; stalled processing messages can be reset after an application-defined threshold. section 11.2
Transactional consumption
A consumer can begin a transaction, call mq.dequeue(1), execute business logic, mark the message completed, and commit. While processing, the message remains invisible to other consumers. section 11.2
Listing 11.1 Creating a queue as a regular Postgres table
CREATE SCHEMA mq; CREATE TYPE mq.status AS ENUM ('new', 'processing', 'completed'); CREATE TABLE mq.queue ( id BIGSERIAL PRIMARY KEY, message JSON NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), status mq.status NOT NULL DEFAULT 'new' );
Inspect the table, JSON payload, timestamp, and new/processing/completed status model. section 11.2
Listing 11.3 Creating a function for retrieving messages from the queue
CREATE OR REPLACE FUNCTION mq.dequeue(messages_cnt INT)
RETURNS TABLE (msg_id BIGINT, message JSON, enqueued_at TIMESTAMPTZ) AS $$
BEGIN
RETURN QUERY
WITH new_messages AS (
SELECT id FROM mq.queue
WHERE status = 'new' ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT messages_cnt
)
UPDATE mq.queue q
SET status = 'processing'
FROM new_messages WHERE q.id = new_messages.id
RETURNING q.id, q.message,
date_trunc('seconds',q.created_at) AS created_at;
END;
$$ LANGUAGE plpgsql;
Inspect the dequeue function that combines FIFO selection, row locking, and status transition. section 11.2
