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
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;
