PostgreSQL'i Message Queue Olarak Kullanmak

PGMQ 

- PGMQ uzantısı kullanılabilir. Burada bazı açıklamalar var

Custom Çözüm

- Burada bir örnek var. Burada başka bir örnek var

1. Claim jobs with FOR UPDATE SKIP LOCKED

Örnek

Şöyle yaparız

SELECT *
FROM queue_job_state
WHERE status='pending'
FOR UPDATE SKIP LOCKED
LIMIT 1;

2. Lease and Fencing Token

Şöyle yaparız

CREATE TABLE queue_job (
    id BIGINT PRIMARY KEY,
    status VARCHAR(20) NOT NULL,
    locked_by VARCHAR(100),
    locked_until TIMESTAMP,
    lease_version BIGINT NOT NULL DEFAULT 0
);

Worker şöyle yapar

WITH candidate AS (
    SELECT id
    FROM queue_job
    WHERE status = 'PENDING'
    ORDER BY id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE queue_job q
SET
    status = 'RUNNING',
    locked_by = 'worker-a',
    locked_until = now() + interval '30 seconds',
    lease_version = q.lease_version + 1
FROM candidate c
WHERE q.id = c.id
RETURNING
    q.id,
    q.payload,
    q.lease_version;

Worker hearbeat extend yapmak isterse şöyle yapar.

UPDATE queue_job
SET locked_until = now() + interval '30 seconds'
WHERE id = 42
  AND lease_version = 7;

Bir başka worker ölen işi almak isterse şöyle yapar

WITH candidate AS (
    SELECT id
    FROM queue_job
    WHERE status = 'RUNNING'
      AND locked_until < now()
    ORDER BY id
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
UPDATE queue_job q
SET
    locked_by = 'worker-b',
    locked_until = now() + interval '30 seconds',
    lease_version = q.lease_version + 1
FROM candidate c
WHERE q.id = c.id
RETURNING
    q.id,
    q.lease_version;

A ve B worker şunu çalıştırır. Bir tanesi başarılı olur.

UPDATE queue_job
SET
    status = 'DONE',
    locked_by = NULL,
    locked_until = NULL
WHERE id = 42
  AND lease_version = 8;