Используйте FOR UPDATE SKIP LOCKED в запросе, который также меняет состояние выбранной строки с ready на processing. Это позволяет параллельным рабочим процессам захватывать разные синтетические строки статей, а сохранённое состояние предотвращает повторный захват после снятия блокировки.

Это руководство предназначено для разработчика, который использует Terminal на Mac с работающим подключением к PostgreSQL. Вы создадите три синтетические статьи, проверите параллельный захват в двух сеансах базы данных и запустите полноценный пример с node-postgres.

Почему два рабочих процесса могут выбрать одну и ту же статью?

Обычный SELECT не резервирует результат. При уровне изоляции Read Committed, используемом в PostgreSQL по умолчанию, два рабочих процесса могут прочитать одну и ту же зафиксированную строку ready до того, как кто-либо из них её изменит. Само по себе объединение такого чтения с последующим обновлением в транзакцию не делает выбор эксклюзивным. PostgreSQL описывает поведение снимков в разделе Изоляция транзакций.

Для захвата нужны две меры защиты: блокировка строки при выборе статьи и изменение её состояния до снятия этой блокировки. PostgreSQL снимает блокировки строк при завершении транзакции. Поэтому отдельное блокирующее чтение с последующей обработкой оставит строку снова доступной после завершения транзакции чтения. См. раздел Явные блокировки.

Во всех рабочих процессах используйте один и тот же протокол захвата. Путь в приложении, который читает статью в состоянии ready и начинает её обработку без успешного захвата, всё ещё может привести к дублированию работы.

Создайте небольшую очередь синтетических статей

Для работы нужны запущенная база данных PostgreSQL, команда psql, доступная в Terminal, и готовая конфигурация подключения. Для раздела JavaScript также установите поддерживаемый выпуск Node.js и npm. В этой статье предполагается, что подключение к базе данных уже работает; установка зависит от настроек вашего Mac.

Функция SKIP LOCKED появилась в PostgreSQL 9.5. Это минимальная версия для данного синтаксиса, а не рекомендация использовать неподдерживаемый выпуск. Используйте поддерживаемую версию PostgreSQL; согласно актуальной полученной документации, текущая версия — PostgreSQL 18. См. примечания к выпуску 9.5 и актуальную документацию по SELECT.

Откройте Terminal и выполните psql с привычными настройками подключения. Откроется интерактивная оболочка для выполнения SQL и просмотра результатов, как описано в документации psql. Выполните приведённую ниже настройку один раз в тестовой базе данных, где таблица article_claim_demo ещё не существует. Команда намеренно не заменяет существующую таблицу.

SHOW server_version;
SHOW transaction_isolation;

CREATE TABLE article_claim_demo (
    id integer PRIMARY KEY,
    title text NOT NULL,
    state text NOT NULL DEFAULT 'ready'
        CHECK (state IN ('ready', 'processing', 'done')),
    rendered_title text
);

INSERT INTO article_claim_demo (id, title) VALUES
    (1, 'Paper shapes'),
    (2, 'Desk arrangements'),
    (3, 'Color combinations');

В примерах предполагается уровень изоляции Read Committed и простая таблица без дополнительных триггеров. Не изменяйте три примера строк, пока не завершите проверку в двух сеансах.

Захватите одну статью одним SQL-оператором

Выполняйте этот оператор каждый раз, когда рабочему процессу нужна статья. Запрос-кандидат блокирует доступную строку; оператор UPDATE регистрирует её захват и возвращает входные поля. В документации PostgreSQL по UPDATE описан этот шаблон с CTE и UPDATE, а также подтверждается, что RETURNING сообщает о фактически обновлённых строках.

WITH candidate AS (
    SELECT id
    FROM article_claim_demo
    WHERE state = 'ready'
    ORDER BY id
    LIMIT 1
    FOR UPDATE SKIP LOCKED
)
UPDATE article_claim_demo AS a
SET state = 'processing'
FROM candidate AS c
WHERE a.id = c.id
  AND a.state = 'ready'
RETURNING a.id, a.title;

Блокировка должна находиться внутри CTE-кандидата. SKIP LOCKED позволяет другому рабочему процессу перейти к следующей строке, если он не может сразу заблокировать кандидата. LIMIT 1 ограничивает каждый захват одной строкой, а соединение по первичному ключу определяет единственную строку для обновления. PostgreSQL прямо описывает блокировку внутри запроса WITH и применение пропускаемых строк для очередей в разделе SELECT: предложение блокировки.

Без явного BEGIN PostgreSQL выполняет оператор в собственной транзакции. При успешном выполнении связанное с выбором обновление и изменение состояния фиксируются до начала дальнейшей обработки. После фиксации блокировка строки снимается, но state = 'ready' исключает эту статью. См. раздел Транзакции.

Обрабатывайте только статью, возвращённую оператором захвата. Пустой результат означает, что эта попытка не получила статью; это не подтверждает, что вся работа завершена. Другие рабочие процессы могут в данный момент удерживать доступные строки. Кроме того, SKIP LOCKED позволяет избежать ожидания блокировки строк, но ожидание блокировок на уровне таблицы всё ещё возможно. Эти ограничения описаны в документации по SELECT.

Проверьте поведение захвата в двух сеансах Terminal

Откройте два окна Terminal и подключитесь через psql к одной базе данных. Начните с исходных трёх строк ready. Следующая последовательность удерживает каждую транзакцию открытой, чтобы вы могли напрямую наблюдать их взаимодействие.

  1. В сеансе A выполните BEGIN ISOLATION LEVEL READ COMMITTED;, затем вставьте полный оператор захвата, приведённый выше. Не завершайте транзакцию. При неизменённых примерах данных возвращённый ID должен быть 1.
  2. В сеансе B выполните BEGIN ISOLATION LEVEL READ COMMITTED;, затем вставьте тот же оператор захвата. Пока сеанс A удерживает ID 1, должен вернуться ID 2.
  3. Выполните ROLLBACK; в сеансе B, затем ROLLBACK; в сеансе A.
  4. В любом сеансе выполните приведённый ниже запрос для проверки. Обе транзакции захвата отменены, поэтому все три статьи снова должны иметь состояние ready.
SELECT id, title, state, rendered_title
FROM article_claim_demo
ORDER BY id;

Это ожидаемые результаты, выведенные из документированного поведения блокировок и отката, а не результаты фактического запуска базы данных. Откат отменяет обновления состояния и снимает блокировки. Если вместо этого выполнить COMMIT, эти статьи останутся в состоянии processing и не будут выбраны при последующих попытках захвата. См. разделы Явные блокировки и Транзакции.

Эта проверка позволяет отдельно наблюдать два полезных свойства: сеанс B переходит к следующей статье, если текущая заблокирована, а зафиксированное изменение состояния не даёт включить статью в будущие выборки. При адаптации существующего рабочего процесса проверьте оба свойства.

Запустите три рабочих процесса с node-postgres

После отката обеих тестовых транзакций создайте в Terminal новый локальный каталог проекта. Установите пакет с помощью документированной команды установки node-postgres:

mkdir article-claim-demo
cd article-claim-demo
npm init -y
npm install pg

Сохраните следующую полноценную программу в файле workers.mjs. Она использует те же переменные окружения для подключения к базе данных, что и psql; общие переменные подключения node-postgres описаны в разделе Подключение. Каждый рабочий процесс пытается захватить одну статью, преобразует синтетический заголовок в верхний регистр и сохраняет результат.

import pg from 'pg';

const { Pool } = pg;
const pool = new Pool({ max: 3 });

const claimSQL = `
WITH candidate AS (
  SELECT id
  FROM article_claim_demo
  WHERE state = 'ready'
  ORDER BY id
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
UPDATE article_claim_demo AS a
SET state = 'processing'
FROM candidate AS c
WHERE a.id = c.id AND a.state = 'ready'
RETURNING a.id, a.title;
`;

async function worker(label) {
  const claimed = await pool.query(claimSQL);

  if (claimed.rows.length === 0) {
    console.log(`${label}: no row available now`);
    return;
  }

  const article = claimed.rows[0];
  const renderedTitle = article.title.toUpperCase();

  const completed = await pool.query(
    `UPDATE article_claim_demo
     SET rendered_title = $2, state = 'done'
     WHERE id = $1 AND state = 'processing'
     RETURNING id`,
    [article.id, renderedTitle]
  );

  if (completed.rowCount !== 1) {
    throw new Error(`${label}: completion did not update one row`);
  }

  console.log(`${label}: completed article ${article.id}`);
}

try {
  const results = await Promise.allSettled([
    worker('A'),
    worker('B'),
    worker('C')
  ]);

  for (const result of results) {
    if (result.status === 'rejected') {
      console.error(result.reason);
      process.exitCode = 1;
    }
  }
} finally {
  await pool.end();
}

Выполните node workers.mjs, затем снова запустите запрос для проверки в psql. Если три готовые статьи не изменялись и другая активность отсутствует, все три состояния должны стать done, а значения в rendered_title должны быть в верхнем регистре. Метки рабочих процессов и порядок вывода в консоль могут не соответствовать порядку статей.

Здесь уместно использовать pool.query, поскольку каждый захват — это один завершённый оператор, а каждое завершение обработки — отдельный оператор. Пул управляет получением и возвратом подключений; pool.end() закрывает пул после завершения попыток. См. раздел Пулинг подключений.

Запрос завершения обработки передаёт значения отдельно с помощью $1 и $2, как описано в разделе Запросы. Программа проверяет, пуст ли массив rows при захвате, и проверяет rowCount, чтобы узнать число обновлённых строк при завершении обработки, как указано в документации pg.Result.

При подготовке в JavaScript была выполнена проверка синтаксиса. В исследовательской среде не было исполняемых файлов PostgreSQL, поэтому SQL и параллельное поведение базы данных сверили с документацией, но здесь не запускали.

Выберите границу транзакции и обработайте незавершённую работу

В примере короткая транзакция захвата фиксируется до преобразования заголовка. Благодаря этому блокировка удерживается недолго, но возникает чёткая граница восстановления: если рабочий процесс остановится после захвата, но до завершения, статья останется в состоянии processing. Повторный запуск программы не захватит её снова.

Для этой демонстрации на компьютере остановите все рабочие процессы и проверьте все статьи processing, прежде чем намеренно снова сделать их доступными. Не сбрасывайте их автоматически, пока предыдущий рабочий процесс ещё может завершиться: в примере нет поля с поколением захвата, которое позволило бы отличить предыдущую попытку от последующей. Одна лишь защита от повторного захвата не обеспечивает автоматическое восстановление или обработку ровно один раз.

Используйте эту схему выбора при адаптации примера:

Ваша задача Подход к транзакции Последствие
Захватить статью, а затем выполнить более длительную работу в приложении Сначала зафиксировать захват одним оператором Сохранить видимое состояние processing и отдельно спроектировать восстановление
Выполнить небольшое преобразование целиком в PostgreSQL Захватить и завершить обработку в одной транзакции Изменения базы данных можно зафиксировать или отменить вместе
Выполнить несколько SQL-операторов в одной транзакции Node.js Получить один клиент из пула и использовать его на всём протяжении Использовать одно и то же подключение для BEGIN, операторов и COMMIT или ROLLBACK

Для последнего подхода node-postgres требует, чтобы все операторы транзакции использовали одного и того же клиента, полученного из пула. Разные вызовы pool.query могут использовать разные подключения и поэтому не могут образовать эту транзакцию. После завершения или отката транзакции освободите клиент. См. разделы Транзакции и Пулинг подключений.

Явно укажите сделанные здесь предположения об уровне изоляции. При уровнях Repeatable Read и Serializable могут возникать ошибки, если строка, которую нужно заблокировать, была изменена с начала транзакции; приложения, использующие такие уровни, должны предусматривать повтор транзакций. PostgreSQL описывает это ограничение в разделе Явные блокировки.

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

Источники