FOR UPDATE SKIP LOCKED supports PostgreSQL work queues with multiple consumers. FOR UPDATE locks the selected rows, while SKIP LOCKED ignores rows another transaction is already processing. section 11.2
A queue consumer typically processes messages as follows:
- Select rows whose status is new, order them by created_at, and limit the batch size. section 11.2
- Lock the selected rows and update their status to processing in one statement. section 11.2
- Let concurrent consumers skip locked rows instead of waiting, allowing them to process different messages in parallel. section 11.2
Dequeuing messages concurrently
A dequeue function can select and claim a batch with: SELECT id FROM mq.queue WHERE status = 'new' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT messages_cnt. An enclosing UPDATE changes those rows to processing and returns their identifiers and payloads. In an explicit transaction, locks remain until COMMIT or rollback; without one, the implicit transaction commits when the function finishes. 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 complete mq.dequeue function, including the CTE, status transition, row limit, and returned message fields. section 11.2
