Почти в каждом приложении есть задача: накапливать записи в таблице и обрабатывать их в фоне. Платежи, уведомления, отчёты — что угодно. Первое решение пишется быстро: расписание, выборка пачки, цикл. Оно даже работает — до первого деплоя с двумя репликами или до первой ошибки во внешней системе.
Эта статья объясняет, что именно ломается и как сделать фоновую обработку надёжно.
Типичный первый подход
Вот код, который встречается в большинстве проектов:
@Component
public class PaymentBatchJob {
private final PaymentRepository repo;
private final PaymentService svc;
@Scheduled(fixedDelay = 300_000)
public void run() {
List<Payment> batch = repo.findTop100ByStatusOrderByCreatedAtAsc(PaymentStatus.NEW);
for (Payment payment : batch) {
try {
svc.process(payment.getId());
} catch (Exception ex) {
log.warn("Failed: {}", payment.getId(), ex);
}
}
}
}
Выборка здесь написана в стиле Spring Data JPA — так этот код и рождается в большинстве проектов. Дальше в статье запросы будут на чистом SQL: захват пачки без FOR UPDATE SKIP LOCKED и RETURNING не сделать, а такие вещи в генерируемые методы репозитория не укладываются. В проекте на jOOQ тот же запрос отдают через dsl.resultQuery(...) — механика от этого не меняется.
Выглядит разумно: раз в 5 минут берём 100 записей со статусом NEW и обрабатываем. Но у этого кода есть несколько серьёзных проблем.
Что здесь ломается
Очередь застревает из-за одной битой записи
Выборка сортируется по дате создания: ORDER BY created_at ASC. Это значит, что при следующем запуске обработчик возьмёт те же самые записи, что и в прошлый раз.
Если один платёж падает с ошибкой каждый раз — он никуда не уходит. Он остаётся в начале очереди и вместе с сотней таких же «застрявших» записей блокирует все новые. Новые платежи накапливаются, но до обработки не доходят.
Счётчика попыток нет. Отложенного повтора нет. Терминального статуса «не вышло» нет.
Зависшие записи после перезапуска
Допустим, внутри svc.process(...) запись сразу переводится в статус PROCESSING — в листинге выше этого не видно, но так обычно и делают, чтобы не взять её повторно. Дальше обработчик идёт во внешнюю систему. Если в этот момент сервис упал — обновление статуса в обратно в NEW уже не происходит. Запись остаётся в PROCESSING навсегда.
Обработчик выбирает только NEW, поэтому такую «зависшую» запись он больше не возьмёт. После каждого перезапуска под нагрузкой таких записей накапливается несколько — и никто о них не знает до жалобы от клиента.
Двойная обработка при двух репликах
Запустили два экземпляра сервиса — оба обработчика читают одни и те же 100 строк одновременно. Один платёж отправится дважды. Для уведомлений это неудобство, для платежей — серьёзный инцидент.
Никакой блокировки выборки нет, и несколько параллельных воркеров договориться между собой не могут.
Один зависший вызов останавливает всё
Обработка идёт последовательно: один платёж → запрос во внешнюю систему → следующий платёж. Если внешняя система «задумалась» на одном запросе — весь цикл встал и ждёт. Таймаутов нет. Остальные 99 платежей стоят в очереди.
Как сделать правильно: очередь задач на PostgreSQL
Для большинства случаев отдельный брокер сообщений не нужен — база данных уже есть, транзакции уже работают. Нужна правильная механика поверх.
Жизненный цикл записи в очереди: из PROCESSING запись возвращается в NEW по ошибке или истёкшей аренде, а когда попытки кончились, уходит вниз в FAILED.
Шаг 1: добавить нужные поля
ALTER TABLE payment
ADD COLUMN attempts int NOT NULL DEFAULT 0,
ADD COLUMN next_attempt_at timestamptz NOT NULL DEFAULT now(),
ADD COLUMN claimed_until timestamptz;
attempts— сколько раз запись брали в работу.next_attempt_at— когда можно попробовать снова (для повторных попыток с задержкой).claimed_until— до какого момента запись «занята» воркером.
ALTER TABLE выше показан в чистом виде. На живой таблице его накатывают с lock_timeout и повтором, иначе одна долгая транзакция подвесит все запросы к таблице платежей: разбор в статье миграции без остановки сервиса.
Одних колонок мало — под них нужны индексы, иначе каждый тик обработчика превратится в полный перебор таблицы платежей. Запрос захвата (он будет ниже) отбирает по двум разным условиям, поэтому и индекса нужно два, причём частичных: в них попадают только те строки, которые обработчик вообще может взять.
CREATE INDEX CONCURRENTLY payment_claim_new_idx
ON payment (next_attempt_at, created_at) WHERE status = 'NEW';
CREATE INDEX CONCURRENTLY payment_claim_stale_idx
ON payment (claimed_until, created_at) WHERE status = 'PROCESSING';
Частичный индекс тут не просто экономия места. Обработанных платежей в таблице со временем станут миллионы, а необработанных — десятки: индекс, который знает только про них, остаётся маленьким и горячим независимо от размера таблицы.
Шаг 2: атомарно захватить пачку
Держать HTTP-вызов внутри транзакции нельзя: соединение с базой занято на всё время ожидания, блокировки на строках висят, а таймаут внешней системы становится таймаутом вашей транзакции. Поэтому захват и обработка должны быть разными транзакциями. Захват — короткая операция, которая атомарно помечает записи как «взяты в работу» прямо в запросе UPDATE:
@Component
@RequiredArgsConstructor
public class PaymentTaskStore {
private final JdbcClient jdbcClient;
@Transactional
public List<UUID> claimBatch(int limit) {
return jdbcClient.sql("""
UPDATE payment SET
status = 'PROCESSING',
attempts = attempts + 1,
claimed_until = now() + interval '5 minutes'
WHERE id IN (
SELECT id FROM payment
WHERE (status = 'NEW' AND next_attempt_at <= now())
OR (status = 'PROCESSING' AND claimed_until < now())
ORDER BY created_at
LIMIT :limit
FOR UPDATE SKIP LOCKED
)
RETURNING id
""")
.param("limit", limit)
.query(UUID.class)
.list();
}
}
Ключевая строка — FOR UPDATE SKIP LOCKED. Это PostgreSQL-конструкция: несколько воркеров одновременно запускают этот запрос, но каждый пропускает уже заблокированные строки и берёт следующие свободные. Двойной обработки не будет — база данных сама распределяет работу между воркерами без какого-либо координатора.
claimed_until решает проблему «зависших» записей: если воркер умер до завершения, через 5 минут условие claimed_until < now() включит эту запись обратно в выборку.
Обратите внимание, что это отдельный бин, а не метод в том же классе, где стоит @Scheduled. Это не вкусовщина. @Transactional в Spring работает через прокси: вызов проходит через обёртку, она открывает транзакцию и передаёт управление дальше. Но обёртка стоит снаружи объекта — если метод с @Transactional вызвать из соседнего метода того же класса, вызов идёт напрямую, минуя прокси, и никакой транзакции не будет. Аннотация при этом на месте, тест «вроде работает», и ошибка всплывает только под нагрузкой. Поэтому всё, что должно выполняться в транзакции, выносят в отдельный бин и вызывают через ссылку на него. Подробнее — в статье @Transactional в Spring.
Ещё одна деталь, которую стоит проговорить: attempts растёт в момент захвата, а не в момент неудачи. То есть это счётчик «сколько раз запись брали в работу», а не «сколько раз обработка провалилась». Разница вылезает, когда воркер умер молча: обработки не было, а попытка засчитана, и до терминального FAILED такая запись доедет быстрее. Считать наоборот — увеличивать счётчик только при явной ошибке — заманчиво, но тогда запись, на которой воркер стабильно падает насмерть, не дойдёт до FAILED никогда и будет крутиться вечно. Из двух зол выбирают первое.
Шаг 3: обработка вне транзакции
Захватили записи, закоммитили — теперь идём во внешнюю систему. HTTP-вызов происходит между транзакциями, а не внутри:
@Component
@RequiredArgsConstructor
public class PaymentBatchJob {
private final PaymentTaskStore store; // ← транзакционные методы живут там
private final LimitsClient limitsClient;
private final Executor executor;
@Scheduled(fixedDelay = 5_000)
public void run() {
List<UUID> claimed = store.claimBatch(100);
claimed.forEach(id -> executor.execute(() -> processOne(id)));
}
private void processOne(UUID paymentId) {
try {
LimitDecision decision = limitsClient.check(paymentId);
store.finishSuccessfully(paymentId, decision);
} catch (Exception ex) {
store.scheduleRetry(paymentId, ex);
}
}
}
Несколько важных изменений по сравнению с первым подходом:
- Записи обрабатываются параллельно через пул потоков, а не последовательно. Один медленный вызов не блокирует остальные.
- Расписание теперь каждые 5 секунд, а не раз в 5 минут — пустой опрос стоит дёшево (один быстрый запрос), а задержка обработки падает с минут до секунд.
- HTTP-клиент
limitsClientдолжен иметь таймауты. Без явного таймаута один зависший вызов займёт поток из пула навсегда.
finishSuccessfully переводит запись в терминальный успешный статус — назовём его DONE — и обнуляет claimed_until. Это обязательно: пока статус остаётся PROCESSING, запись подходит под условие status = 'PROCESSING' AND claimed_until < now(), и через пять минут её захватит следующий тик. Обработали и не сменили статус — считайте, что не обработали.
И главное, чего в этом листинге не хватает. run() складывает 100 задач в executor и тут же возвращается, через 5 секунд складывает ещё 100. Если обработка медленнее, чем захват, очередь исполнителя растёт без предела — а аренда claimed_until тикает и у тех задач, которые ещё лежат в ней и не начинались. Через пять минут аренда протухнет, соседняя реплика подберёт ту же запись — и мы получим ровно ту двойную обработку, от которой весь этот раздел.
Лечится это ограниченной очередью: пул с конечной вместимостью и политикой CallerRunsPolicy.
@Bean
Executor paymentExecutor() {
var executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(8);
executor.setMaxPoolSize(8);
executor.setQueueCapacity(100); // не безразмерная
executor.setRejectedExecutionHandler(new CallerRunsPolicy()); // очередь полна — тик ждёт
executor.initialize();
return executor;
}
Когда очередь заполнена, CallerRunsPolicy заставляет выполнить задачу тот поток, который её отправлял, — то есть поток планировщика. Тик замедляется ровно настолько, насколько не справляется обработка, и новых записей не захватывает. Это и есть обратная связь: система сама притормаживает вместо того, чтобы копить работу, у которой истекает аренда.
Шаг 4: повторные попытки с задержкой и терминальный отказ
Когда обработка падает — запись не остаётся в начале очереди. Она уходит в конец с задержкой:
// в том же PaymentTaskStore — иначе @Transactional снова не сработает
@Transactional
public void scheduleRetry(UUID paymentId, Exception ex) {
Payment payment = loadForUpdate(paymentId); // читаем под блокировкой
if (payment.attempts() >= MAX_ATTEMPTS) {
payment.markFailed(ex.getMessage()); // терминальный статус FAILED
} else {
payment.scheduleNextAttempt(backoff(payment.attempts())); // возврат в NEW с задержкой
}
save(payment);
}
Запись здесь читают под блокировкой (SELECT ... FOR UPDATE) и в той же транзакции сохраняют. Без этого метод читает и пишет вразнобой: пока он думает, ту же запись успевает подобрать другой воркер по истёкшей аренде, и одно из двух изменений пропадёт.
backoff — функция, которая увеличивает паузу с каждой попыткой: например, 1 мин → 2 мин → 4 мин → 8 мин. После N попыток запись получает статус FAILED с причиной ошибки — это сигнал человеку разобраться вручную. Очередь не застревает.
И обязательный разброс к этим паузам. Внешняя система упала на пять минут, за это время сто записей получили ошибку — и все сто встали на повтор в одну и ту же секунду. Когда система поднимется, она получит сто одновременных запросов вместо ровного потока, и, скорее всего, ляжет снова. Лечится случайной добавкой к паузе:
private Instant nextAttemptAt(int attempts) {
Duration base = Duration.ofMinutes(1L << Math.min(attempts, 6)); // 1, 2, 4 … до 64 минут
long jitterMs = ThreadLocalRandom.current().nextLong(base.toMillis() / 2);
return Instant.now().plus(base).plusMillis(jitterMs); // ±50 % разброса
}
Половина базовой паузы как разброс — разумное умолчание: сто записей растянутся на интервал, а не ударят пачкой. Верхний предел (1L << 6 — около часа) нужен, чтобы пауза не выросла до суток на седьмой попытке. И обратите внимание: разброс считается при планировании, а не при выборке, — иначе записи снова синхронизируются.
Что делать с записями в статусе FAILED
Терминальный статус — не конец истории, а начало отдельного процесса, и без него FAILED превращается в кладбище, куда никто не смотрит.
Первое: их надо видеть. Число записей в FAILED за окно — метрика на графике (см. раздел «Глубже» про наблюдаемость ниже), а не строка в таблице, которую кто-то однажды откроет. Рост — сигнал, что сломалось не «одно сообщение», а что-то общее.
Второе: причина должна быть читаемой. В записи хранят не только текст исключения, но и вид ошибки — это то, что позволяет разбирать их пачками, а не по одной:
ALTER TABLE payment_task
ADD COLUMN failure_kind text, -- 'EXTERNAL_5XX' | 'VALIDATION' | 'NOT_FOUND' | 'UNKNOWN'
ADD COLUMN failure_detail text,
ADD COLUMN failed_at timestamptz;
Разделение простое: наша ошибка (неверные данные, отсутствующая сущность — повторять бессмысленно, нужен разбор или исправление данных) и чужая ошибка (внешняя система отказала окончательно — повторять после её починки осмысленно). Без этого признака дежурный не знает, что делать со списком.
Третье: возврат в работу — обычная операция, а не правка базы руками. После починки внешней системы записи с видом EXTERNAL_5XX возвращают в очередь одной операцией:
UPDATE payment_task
SET status = 'PENDING', attempts = 0, next_attempt_at = now(),
failure_kind = NULL, failure_detail = NULL
WHERE status = 'FAILED'
AND failure_kind = 'EXTERNAL_5XX'
AND failed_at > now() - interval '1 day';
Этот запрос стоит завернуть в служебную ручку или команду с правами, а не выполнять из клиента базы: тогда у операции есть журнал и её может сделать дежурный, а не только тот, у кого есть доступ к базе. И обязательно с фильтром — возвращать все FAILED за всю историю опасно: среди них есть те, что упали по своей вине, и они упадут снова.
Четвёртое: автоматический возврат — можно, но осторожно. Если внешняя система восстанавливается сама (а это обычный случай), разумно завести задачу, которая раз в час возвращает в работу записи с чужой ошибкой, не более N за проход и не более M раз на запись (чтобы не крутить вечно). Это снимает большую часть ручной работы; наши ошибки не возвращают никогда автоматически.
Пятое: экран для человека. Список FAILED с фильтром по виду и причине, с кнопкой «вернуть в работу» и с показом самого сообщения. Это не роскошь: без него разбором занимается разработчик с доступом к базе, то есть дорогой человек по любому поводу.
И срок хранения. Записи FAILED, разобранные и неактуальные, чистят по возрасту (обычно месяц-два) — иначе таблица очереди растёт вечно, и выборка по ней постепенно замедляется.
Идемпотентность внешнего вызова
Надёжная очередь гарантирует, что задача не потеряется, — и ничего не говорит о том, что она не выполнится дважды. А выполнится: таймаут на вызове означает «ответ не пришёл», а не «действие не произошло». Внешняя система могла принять платёж и не успеть ответить; повтор спишет деньги второй раз.
Это не крайний случай, а обычный: любой сбой сети на длинном запросе выглядит именно так. Поэтому у надёжной очереди задач есть обязательный спутник — ключ идемпотентности на внешнем вызове.
Как это работает. Мы передаём внешней системе ключ, который не меняется между повторами, и она обязуется при повторе с тем же ключом вернуть результат первой попытки, а не выполнить операцию заново:
private void processOne(UUID paymentId) {
PaymentTask task = store.byId(paymentId);
try {
// Ключ выводится из задачи, а не генерируется заново при каждой попытке
LimitDecision decision = limitsClient.check(paymentId, task.idempotencyKey());
store.finishSuccessfully(paymentId, decision);
} catch (Exception ex) {
store.scheduleRetry(paymentId, ex);
}
}
Откуда берётся ключ — самое важное. Он должен быть детерминированным: один и тот же для всех попыток одной задачи. Три рабочих варианта:
- Идентификатор задачи — проще всего, если он уже уникален:
payment-task-<id>. - Составной из смысла операции:
payment-<paymentId>-attempt-limit-check. Годится, когда по одной сущности выполняется несколько разных операций. - Сгенерированный один раз и сохранённый в записи — при создании задачи пишем ключ в колонку и потом используем его. Нужен, когда естественного ключа нет.
Чего делать нельзя: генерировать ключ в момент вызова (тогда у каждой попытки он свой, и идемпотентности нет — а код при этом выглядит правильно) и использовать время (то же самое).
Если внешняя система ключей не поддерживает. Бывает, и тогда есть три пути. Спросить состояние перед повтором: «проведён ли платёж с таким идентификатором?» — если у системы есть такой запрос, это полноценная замена. Сверка постфактум: регулярно сравнивать свои записи с их выпиской и разбирать расхождения — дорого, но иногда единственный вариант; для денег он обязателен независимо от всего остального. Не повторять автоматически: задача при таймауте уходит в FAILED со особым видом («неизвестно, выполнилось ли») и разбирается человеком. Последнее звучит слабо, но для необратимых операций без ключа это честнее автоматического повтора.
И своя сторона тоже. Обработчик задачи должен переносить повтор внутри себя: если он пишет в свою базу, запись должна быть по ключу задачи (вставка с проверкой уникальности), а не «добавить строку». Иначе двойная обработка создаст два резерва по одной задаче, и виновата будет уже не внешняя система.
Правило, в которое это сворачивается: очередь отвечает за «не потеряется», ключ идемпотентности — за «не удвоится». Нужны оба; одна без другой даёт либо потерянные задачи, либо двойные списания.
Размер пачки и размер пула
В коде выше стоят два числа — claimBatch(100) и пул исполнителей неизвестного размера, — и взяты они с потолка. Связь между ними есть, и считается она от внешней системы, а не от красоты.
Что задаёт потолок. Почти всегда — внешняя система: её лимит запросов в секунду или её способность отвечать. Отсюда обратный порядок расчёта: сначала узнаём, сколько она выдерживает, и только потом выбираем свои числа.
Размер пула — от пропускной способности и времени ответа. Работает закон Литтла: чтобы выдавать λ запросов в секунду при времени ответа W, нужно λ × W одновременных запросов. Внешняя система разрешает 50 запросов в секунду и отвечает за 200 мс — значит, одновременных нужно 50 × 0,2 = 10. Пул из десяти потоков, и не больше: двадцать потоков просто получат отказы по лимиту.
Размер пачки — от интервала запуска и пропускной способности. Задание идёт раз в 5 секунд, обрабатываем 50 задач в секунду — значит, за интервал успеваем около 250. Брать пачку заметно больше бессмысленно: лишние задачи будут ждать в памяти, удерживая аренду, и при перезапуске вернутся в очередь. Брать заметно меньше — простаивать. Разумное правило: пачка примерно равна тому, что успеваете обработать за интервал, с небольшим запасом.
Число копий сервиса умножает, а не делится. Важная деталь: если сервис работает в трёх копиях, каждая возьмёт свою пачку и будет держать свой пул — то есть нагрузка на внешнюю систему в три раза больше расчётной. Значит, считать надо на все копии: либо делить пул на число копий, либо ставить общий ограничитель темпа (счётчик во внешнем хранилище, см. ограничение частоты). Забытая тройка здесь — самая частая причина «мы же считали, а их лимит выбило».
Что ещё ограничивает, кроме внешней системы. Пул соединений к своей базе: каждая задача в обработке рано или поздно пишет результат, и если пул исполнителей больше пула соединений, потоки встанут в очередь за соединением. Разумно держать пул исполнителей не больше пула соединений минус запас на обычные запросы. И память: пачка из десяти тысяч записей с полезной нагрузкой — это уже заметный объём в куче.
Как проверить, что числа выбраны верно. Три наблюдения: время обработки пачки должно быть меньше интервала запуска (иначе задания накладываются); возраст самой старой необработанной записи не должен расти (иначе не успеваем — тогда либо чаще, либо больше копий, либо разговор с внешней системой про лимит); доля отказов по лимиту должна быть близка к нулю (иначе пул велик). Все три — из тех метрик, которые разобраны в разделе «Глубже» ниже.
Ориентиры на старт, чтобы не начинать с нуля: интервал 5 секунд, пачка 100, пул 10, и дальше подгонять по трём наблюдениям выше. Это заведомо безопасные числа для внешней системы с лимитом в десятки запросов в секунду.
Когда нужен фреймворк пакетной обработки
Spring Batch — инструмент для конечных задач: выгрузить реестр платежей за месяц в файл, мигрировать десять миллионов строк, перетарифицировать всех клиентов раз в квартал.
Его ключевая возможность — перезапуск с места сбоя: «упало на 7-м миллионе — перезапустили, продолжило с 7-го». Для этого он хранит состояние прогона и поддерживает политики пропуска и повтора на уровне порций данных.
Рядом часто называют Quartz и Celery с beat, но это инструменты другой природы, и путать их со Spring Batch не стоит. Quartz — планировщик: он умеет хранить расписание в базе, не дать двум узлам запустить одну задачу дважды и разобраться с пропущенными запусками, пока сервис лежал. Celery beat в мире Python — то же самое, отправка периодических задач в очередь по расписанию. Ни тот, ни другой не продолжат обработку с седьмого миллиона: они решают вопрос «когда запускать», а не «как дожать до конца». Перезапуск с места сбоя — только у Spring Batch.
Для нашей задачи (непрерывная очередь входящих платежей) такой фреймворк избыточен: поток не имеет конца, перезапуск с места не нужен, а вопросы параллельного захвата и повторных попыток остаются теми же — фреймворк их не решает.
Простое правило: конечный пакет с перезапуском → фреймворк; бесконечный поток задач → очередь задач на PostgreSQL.
Когда добавлять брокер сообщений
Брокер (RabbitMQ, Kafka) решает те же задачи, что и очередь на PostgreSQL, но добавляет новые возможности: несколько независимых типов потребителей одного события, масштабирование воркеров без нагрузки на основную базу данных, встроенные очереди недоставленных сообщений.
Переходить к брокеру имеет смысл, когда объёмы вырастают до десятков тысяч задач в минуту или когда одно событие должны обрабатывать несколько разных сервисов. До этого — брокер лишняя инфраструктура с эксплуатационными расходами.
И вторая половина этой мысли, без которой порог по объёму читается неверно: очередь задач и брокер решают разные задачи, а не одну и ту же с разной производительностью.
Очередь задач — это «исполнитель и работа». Одна задача выполняется ровно одним исполнителем; после выполнения она закрыта. Смысл конструкции — надёжно раздать работу между копиями сервиса, повторить при сбое, довести до конца. Ключевые свойства: у задачи есть состояние и число попыток, её можно посмотреть запросом, поправить и вернуть в работу, а история остаётся в таблице.
Брокер — это «событие и подписчики». Одно событие читают несколько независимых потребителей, каждый со своей скоростью и своей целью: один пишет в витрину, второй отправляет письмо, третий наполняет аналитику. Добавить четвёртого не требует менять издателя — и вот это свойство таблицей задач не воспроизводится: чтобы дать ту же работу двум разным обработчикам, придётся создавать две записи, то есть издатель начинает знать про подписчиков.
Отсюда практическая развилка, которая честнее порога по объёму:
| Вопрос | Ответ |
|---|---|
| Одну работу выполняет один исполнитель, надо надёжно и с повторами | Очередь задач на базе |
| Одно событие нужно нескольким независимым получателям | Брокер |
| Нужно переиграть поток заново (новый потребитель, пересчёт витрины) | Брокер с хранением |
| Нужно посмотреть и поправить конкретную задачу запросом | Очередь задач на базе |
| Нужна транзакционность с бизнес-данными (одна транзакция на запись и задачу) | Очередь задач на базе |
| Десятки тысяч сообщений в минуту и выше | Брокер |
Из таблицы видно, что переход на брокер по объёму — только одна из шести причин, и обычно не первая. Чаще настоящая причина — «появился второй получатель того же события», и она приходит раньше объёмов.
И схема, которая объединяет оба, — самая частая в живых системах: таблица исходящих сообщений плюс брокер. Событие пишется в таблицу в той же транзакции, что бизнес-данные (транзакционность от очереди задач), а отправщик перекладывает его в брокер (несколько потребителей от брокера). То есть выбор не «или-или», и про механику — в статье про паттерны интеграции.
Глубже: ровно один раз в сутки: ShedLock и advisory lockрасширенное
FOR UPDATE SKIP LOCKED раздаёт очередь между репликами, и это решает «много задач, много воркеров». Задачу «сформировать отчёт в три ночи ровно один раз» он не решает: очереди нет, есть расписание, и три реплики с @Scheduled запустят его трижды.
ShedLock. Библиотека для того же Spring @Scheduled: перед запуском задача пытается записать строку в таблицу shedlock (name, lock_until, locked_by); удалось, значит выполняет, не удалось, значит другая реплика уже взяла. lockAtMostFor задаёт, когда замок истечёт сам, если реплика умерла посреди задачи, lockAtLeastFor не даёт второй реплике с чуть отстающими часами запустить задачу сразу после первой. Провайдер замка это та же база через JdbcTemplateLockProvider, лишней инфраструктуры нет. Как настройки ShedLock связаны с остановкой пода, разбирает статья про планировщик и корректное завершение.
Advisory lock. Тот же эффект средствами PostgreSQL: SELECT pg_try_advisory_lock(hashtext('nightly-report')) возвращает истину только одному соединению, остальные получают ложь и выходят; замок живёт до конца сессии или транзакции (pg_try_advisory_xact_lock), поэтому смерть реплики отпускает его сама. Без таблицы и без библиотеки, но замок держится на одном соединении из пула, и его нельзя отпускать раньше, чем задача закончена.
Что общего и что выбрать. Оба дают «не больше одного запуска одновременно», но не «ровно один раз»: реплика, умершая посередине, оставляет задачу недоделанной, и повторный запуск обязан быть идемпотентным, как и в очереди выше (отчёт за день заменяется целиком, не дописывается). ShedLock понятнее и виден в таблице, advisory lock дешевле и без зависимостей; при Kubernetes ту же задачу честнее отдать CronJob, который запускает один под, а не реплики сервиса, о чём статья про Job и CronJob.
Глубже: наблюдаемость очереди задачрасширенное
Все ветки выше заканчиваются «сигнал человеку разобраться», и стоит назвать, по каким числам он приходит. У очереди на таблице их четыре, и все считаются одним запросом на расписании и отдаются как метрики.
Возраст самой старой необработанной задачи, now() - min(created_at) where status = 'PENDING': главная метрика, потому что она в секундах и понятна без контекста; тревога на «старше пяти минут» для платежей и «старше часа» для отчётов. Число задач по статусам (PENDING, CLAIMED, FAILED) в динамике: растущий PENDING при постоянных воркерах это нехватка мощности, растущий CLAIMED это зависшие аренды, а любой рост FAILED это тревога с примером причины из последней строки. Доля повторных попыток: attempts > 1 среди обработанных за час, по ней деградацию внешней системы видно раньше отказа. И длительность обработки одной задачи с процентилями через Timer вокруг обработчика.
Что смотреть на панели рядом: пул соединений (воркеры и HTTP делят один), и ошибки внешней системы, к которой ходят задачи. Как та же тройка выглядит для потребителей брокера и outbox, разбирает статья про метрики в разделе наблюдаемости.
Коротко
- Простой
@Scheduled-цикл ломается при параллельных репликах, зависших записях и последовательных HTTP-вызовах без таймаутов. Правильный захват задач — атомарный UPDATE сFOR UPDATE SKIP LOCKED: база сама распределяет работу между воркерами. claimed_until(«аренда задачи») защищает от зависших записей после перезапуска — просроченную аренду воркеры подберут автоматически. HTTP-вызовы — между транзакциями, не внутри; обработка — параллельная через пул, не последовательная.- Повторные попытки с нарастающей задержкой и терминальный статус
FAILEDне дают очереди застрять. Фреймворк пакетной обработки — для конечных задач с перезапуском; брокер — для масштаба и нескольких потребителей. - Задачу по расписанию ровно один раз на кластер держат ShedLock (строка-замок в базе с
lockAtMostFor) или advisory lock PostgreSQL; оба не гарантируют завершение, повтор обязан быть идемпотентным; в Kubernetes честнее CronJob. - Очередь наблюдают по возрасту самой старой необработанной задачи (тревога в секундах), числу по статусам, доле повторов и длительности обработки.
- К растущим паузам обязателен разброс (±50 % базовой паузы) и верхний предел, иначе сто записей после аварии внешней системы повторятся в одну секунду и уронят её снова.
- У
FAILEDесть вид ошибки (наша или чужая), метрика на графике, операция возврата в работу с фильтром и правами, возможный автоматический возврат для чужих ошибок и срок хранения. - Очередь отвечает за «не потеряется», ключ идемпотентности — за «не удвоится»: ключ выводят из задачи, а не генерируют при вызове; без поддержки ключей остаются запрос состояния, сверка постфактум или ручной разбор.
- Пул считают от лимита внешней системы (темп × время ответа), пачку — от интервала запуска, и умножают на число копий сервиса; проверяют по трём наблюдениям: время пачки меньше интервала, возраст старейшей записи не растёт, отказов по лимиту нет.
- Очередь задач и брокер решают разное: одна работа одному исполнителю с историей и правкой против одного события нескольким независимым получателям; в живых системах их совмещают таблицей исходящих плюс брокером.
Что почитать дальше
- Паттерны отказоустойчивости — таймауты, повторные попытки, автоматический выключатель для вызова внешних систем.
- Распределённые паттерны — transactional outbox и идемпотентность при работе с брокером.
@Transactionalв Spring — почему границы транзакций важны и где декларативное управление не срабатывает.- Spring Scheduled, Async, виртуальные потоки: интервал тика, общий поток планировщика и ShedLock, чтобы обработчик запускала одна реплика.