Очереди на PostgreSQL: как применять SKIP LOCKED

Если два воркера одновременно выбирают задачу из таблицы PostgreSQL, обычный SELECT FOR UPDATE заставит одного ждать другого. SKIP LOCKED меняет поведение: занятые строки пропускаются, а каждый воркер может забрать свою пачку. Ниже — схема для PostgreSQL 16, в которой захват задачи занимает короткую транзакцию, а сама работа выполняется уже после её завершения.

Что именно делает SKIP LOCKED

FOR UPDATE блокирует выбранные строки до конца транзакции. Другой запрос, который пытается заблокировать те же строки, обычно ждёт. Если добавить SKIP LOCKED, запрос не ждёт занятые строки, а ищет следующие доступные.

Пропуск заблокированной строки — это не пропуск задачи навсегда: запрос просто не отдаёт её этому воркеру сейчас. Когда блокировка снимется, задача снова сможет попасть в выборку, если её условиям отбора по-прежнему соответствует.

SKIP LOCKED решает гонку за строку, но не гарантирует, что задача выполнится ровно один раз. Это отдельная ответственность приложения.

Механизм рассчитан на очереди и похожие задачи, где несколько потребителей конкурируют за записи. Результат не является согласованным снимком всей таблицы: часть строк может быть пропущена именно из-за блокировок. Поэтому применять SKIP LOCKED для обычного пользовательского списка, где важна полнота выдачи, не стоит.

Атомарно забираем пачку задач

Пример рассчитан на PostgreSQL 16. Предположим, у очереди есть состояния ready, running и done, а отложенную задачу можно запускать после значения run_at. Воркеры получают задания пакетами, чтобы реже обращаться к базе.

CREATE TABLE jobs (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    status text NOT NULL DEFAULT 'ready'
        CHECK (status IN ('ready', 'running', 'done', 'failed')),
    priority integer NOT NULL DEFAULT 0,
    run_at timestamptz NOT NULL DEFAULT now(),
    payload jsonb NOT NULL,
    locked_by text,
    locked_at timestamptz,
    attempts integer NOT NULL DEFAULT 0
);

CREATE INDEX jobs_ready_pick_idx
    ON jobs (priority DESC, run_at, id)
    WHERE status = 'ready';

Частичный индекс содержит только готовые к запуску записи. Он может ускорить выборку, но конкретный план зависит от размера таблицы и распределения данных: проверяйте его на реальных данных через EXPLAIN (ANALYZE, BUFFERS).

Захват задач и перевод их в running выполняются одним запросом:

WITH picked AS (
    SELECT id
    FROM jobs
    WHERE status = 'ready'
      AND run_at <= now()
    ORDER BY priority DESC, run_at, id
    LIMIT 10
    FOR UPDATE SKIP LOCKED
)
UPDATE jobs AS j
SET status = 'running',
    locked_at = clock_timestamp(),
    locked_by = $1,
    attempts = attempts + 1
FROM picked
WHERE j.id = picked.id
RETURNING j.id, j.payload, j.attempts;

Параметр $1 — идентификатор воркера, переданный через механизм параметризованных запросов вашего драйвера. Не подставляйте его в SQL конкатенацией. После получения результата приложение должно зафиксировать транзакцию, а затем обрабатывать payload. Если задача выполняется внутри открытой транзакции, блокировки остаются активны всё это время.

1. Откройте транзакцию. Выполните запрос захвата от имени одного воркера. Проверка: запрос вернул не больше десяти строк и у них статус уже running.
2. Зафиксируйте транзакцию сразу после захвата. Не держите соединение с открытой транзакцией, пока отправляете письмо или обрабатываете файл. Проверка: другой воркер может продолжить забирать следующие готовые задачи.
3. Выполните работу вне транзакции. На успехе отметьте задачу завершённой отдельным коротким запросом. Проверка: строка получила статус done.
4. Обработайте сбой отдельно. Запишите ошибку и решите, повторять задачу или переводить её в failed. Проверка: после перезапуска воркера запись не остаётся бесконечно в running.

Три варианта захвата: цена ожидания и восстановления

Для большинства фоновых задач практичнее второй вариант: база быстро раздаёт работу, не оставаясь в транзакции во время внешних операций. Цена — нужно явно продумать восстановление после падения процесса. Поле locked_at помогает находить зависшие записи, но само по себе не делает повторную обработку безопасной.

Что ломается при реальной нагрузке

Воркер упал после захвата

Запись уже имеет статус running, но работу никто не завершит. Нужен процесс восстановления: например, периодически находить задачи, которые слишком долго не обновляли отметку активности, и возвращать их в очередь. Порог нельзя выбрать наугад: он должен учитывать максимальную длительность штатной задачи. Если работа может быть долгой, воркеру понадобится продлевать аренду или подавать heartbeat.

Повторное выполнение меняет результат

Процесс может завершить внешний вызов, например списание или отправку сообщения, и упасть до того, как записал done. После восстановления задача запустится снова. Поэтому обработчик должен переносить повтор: использовать ключ идемпотентности, проверять уже созданный результат или иначе защищать побочный эффект. Блокировка строки не может объединить транзакцию PostgreSQL с операцией во внешнем сервисе.

Порядок задач не строгий

ORDER BY задаёт порядок среди доступных строк, но воркер пропустит задачу с высоким приоритетом, если её уже заблокировали. В результате более поздняя запись может начать выполняться раньше. Если бизнес-правило требует строгой последовательности, например для операций одного аккаунта, одной общей очереди с SKIP LOCKED может быть недостаточно: нужно отдельно задавать границы параллелизма и порядок для этой сущности.

Индекс не совпадает с фильтром

Без подходящего индекса PostgreSQL может просматривать много строк в поиске небольшого числа готовых задач. Начните с индекса по условиям и сортировке, затем посмотрите план запроса и время на данных, похожих на production. Частые смены статуса и удаление старых задач тоже создают нагрузку на таблицу и индексы; за ростом размера и частотой очистки нужно следить отдельно.

Инвалидация и архивирование старых записей — отдельная задача; здесь важен только захват доступных строк.

Что делать дальше

Запустите два и более воркера на тестовой очереди и проверьте, что одна и та же строка не выдаётся им одновременно. Затем принудительно завершите один процесс после захвата: задача должна обнаружиться механизмом восстановления, а повторная обработка — не создавать второй побочный эффект. В production смотрите на возраст самой старой задачи в ready, число записей в running, время обработки и количество повторных попыток. Эти показатели покажут не только скорость выдачи, но и случаи, когда очередь перестала продвигаться.