Job queue bằng Postgres — FOR UPDATE SKIP LOCKED

Mấy dự án nhỏ của mình hay có việc nền: gửi email xác nhận, đồng bộ đơn hàng, resize ảnh. Đủ chậm để không chạy ngay trong request, mà chưa đủ lớn để đáng dựng thêm cụm Redis hay RabbitMQ rồi phải trực nó lúc 2 giờ sáng. Nên mình nhét luôn hàng đợi job vào PostgreSQL — database vốn đã nằm đó, có backup, có monitoring.

Điểm mấu chốt để chuyện này chạy đúng nằm ở hai chữ: SKIP LOCKED.

Cách làm sai mà ai cũng từng làm

Bản đầu tiên của mình trông rất hợp lý: một bảng jobs có cột status, worker chạy vòng lặp, SELECT job pending cũ rồi UPDATE nó sang running.

const { rows } = await pool.query(
  "SELECT * FROM jobs WHERE status = 'pending' ORDER BY id LIMIT 1"
);
await pool.query("UPDATE jobs SET status = 'running' WHERE id = $1", [rows[0].id]);

Chạy một worker thì êm. Chạy ba worker thì bắt đầu có job bị xử lý hai lần: cả ba cùng SELECT ra đúng một dòng, cùng thấy nó pending, rồi cùng UPDATE. Khách nhận hai email xác nhận, hoặc tệ hơn — tiền bị trừ hai lần.

Cách chữa theo phản xạ là SELECT ... FOR UPDATE. Đúng hơn, nhưng worker thứ hai bị chặn, phải chờ worker thứ nhất xong mới biết dòng đó không còn dành cho mình. Mười worker thì mười cái chờ nhau — còn chậm hơn chạy một worker.

SKIP LOCKED: khoá mỗi dòng, không khoá hàng đợi

FOR UPDATE SKIP LOCKED là FOR UPDATE cộng thêm một câu: dòng nào đang bị transaction khác giữ thì bỏ qua, đừng chờ. Nghĩa là mỗi worker lấy được một job khác nhau, ngay lập tức, không ai chặn ai.

Và điều hay là mình không cần SELECT rồi mới UPDATE — gộp cả hai vào một câu, để việc "giành được job" và "đánh dấu job" là một thao tác nguyên tử:

import { Pool } from "pg";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

type Job = {
  id: number;
  kind: string;
  payload: unknown;
  attempts: number;
};

const CLAIM_SQL = `
  UPDATE jobs
  SET status     = 'running',
      attempts   = attempts + 1,
      locked_at  = now(),
      locked_by  = $1
  WHERE id = (
    SELECT id FROM jobs
    WHERE status = 'pending'
       OR (status = 'running' AND locked_at < now() - interval '5 minutes')
    ORDER BY priority DESC, id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
  )
  RETURNING id, kind, payload, attempts;
`;

async function claim(workerId: string): Promise<Job | null> {
  const { rows } = await pool.query<Job>(CLAIM_SQL, [workerId]);
  return rows[0] ?? null;
}

Để ý điều kiện thứ hai: job running mà locked_at đã cũ hơn 5 phút thì được coi là của worker đã chết và cho phép giành lại. Đây là visibility timeout, phiên bản đơn giản nhất của heartbeat. Không có nó thì một worker bị kill giữa đường là job đó treo vĩnh viễn.

Vòng lặp worker gọn như vậy:

async function runWorker(workerId: string): Promise<void> {
  for (;;) {
    const job = await claim(workerId);
    if (!job) {
      await new Promise((r) => setTimeout(r, 1000)); // hết việc thì thở 1 giây
      continue;
    }

    try {
      await handle(job);
      await pool.query(
        "UPDATE jobs SET status = 'done', finished_at = now() WHERE id = $1",
        [job.id]
      );
    } catch (err) {
      // attempts đã tăng lúc claim rồi, quá 3 lần là bỏ vào dead letter
      const next = job.attempts >= 3 ? "dead" : "pending";
      await pool.query(
        "UPDATE jobs SET status = $2, last_error = $3 WHERE id = $1",
        [job.id, next, String(err)]
      );
    }
  }
}

Bảng và index đi kèm — index partial chỉ trên dòng pending, vì đó là tập dữ liệu nhỏ còn done thì phình ra theo thời gian:

CREATE TABLE jobs (
  id         bigserial PRIMARY KEY,
  kind       text NOT NULL,
  payload    jsonb NOT NULL DEFAULT '{}',
  status     text NOT NULL DEFAULT 'pending',
  priority   int  NOT NULL DEFAULT 0,
  attempts   int  NOT NULL DEFAULT 0,
  locked_at  timestamptz,
  locked_by  text,
  last_error text,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX jobs_queue_idx ON jobs (priority DESC, id)
  WHERE status = 'pending';

Mấy chỗ mình vấp thật

Đừng gọi API ngoài trong lúc giữ khoá. Mình từng để hàm handle() chạy ngay bên trong transaction claim job. Payment gateway treo 30 giây là transaction giữ khoá 30 giây, và vài chục worker thì pool connection cạn. Nguyên tắc: claim xong, commit, rồi mới làm việc nặng.

Job dài phải có heartbeat. Con số 5 phút ở trên là timeout chết. Một job import file mất 12 phút sẽ bị worker khác giành lại giữa lúc worker chính vẫn đang chạy — job bị xử lý hai lần, lần này không phải do race mà do timeout đặt sai. Cách mình làm: trong lúc xử lý, cứ 60 giây chạy UPDATE jobs SET locked_at = now() WHERE id = $1 một lần để gia hạn.

Polling mãi thì DB phải làm việc không. Một worker hỏi 1 giây một lần là 86.400 query rỗng mỗi ngày, nhân lên số worker. Mình dùng LISTEN/NOTIFY: lúc insert job thì NOTIFY, worker LISTEN và chỉ tỉnh dậy khi có việc. Polling vẫn giữ làm lưới an toàn, nhưng giãn ra 30 giây.

Bảng done cần dọn. Đây là chỗ dễ quên nhất, và là lý do bảng phình dù bạn đã xoá sạch dòng cũ. Job xong thì archive hoặc xoá theo lô định kỳ, chứ để vài chục triệu dòng done chỉ để tra lịch sử thì index nào cũng chậm dần.

Khi nào thì nên đổi

Postgres làm hàng đợi tốt hơn mình tưởng, nhưng có ranh giới: vài nghìn job một phút, một database, không cần fan-out cho nhiều consumer group thì thừa sức. Vượt mức đó — hàng trăm nghìn message mỗi giây, cần replay, cần đẩy cho nhiều hệ thống — thì Kafka hay RabbitMQ vẫn đúng hơn, đừng cố nhồi. Còn với đa số hệ thống SME, một bảng jobs cộng SKIP LOCKED là đủ: hàng đợi bền, có transaction, backup sẵn, không thêm service nào phải trực.

Nếu hệ thống của bạn cũng đang ghi DB rồi mới publish event, cách mình xử lý nằm ở bài Outbox Pattern — cùng một ý tưởng: để database làm nhân chứng.

Hàng đợi của bạn đang nằm ở đâu, và có ai từng bị khách gọi vì nhận hai email xác nhận chưa?

📋 Phụ lục thuật ngữ