AsyncWorker

ru.bgcrm.context.async.AsyncWorker — штатная точка запуска короткоживущих фоновых задач внутри процесса ERP. Он создаёт контекст пользователя, управляет транзакцией, переносит MDC, журналирует результат и публикует телеметрию.

Новый код не должен создавать Thread или executor через java.util.concurrent.Executors: архитектурные правила требуют использовать AsyncWorker.

AsyncWorker не является надёжной очередью. Задачи хранятся только в памяти текущего инстанса, теряются при остановке JVM и не переносятся на другой инстанс. Повтор (submitWithRetry) и дедупликация (submitUnique) тоже работают только в памяти одного инстанса. Для гарантированного исполнения храните состояние операции в БД и используйте модуль Task либо планировщик.

Модель выполнения

Executor создаёт отдельный виртуальный поток async-worker-pool-* для каждой задачи. Общего ограничения числа одновременно запущенных задач и очереди с backpressure у AsyncWorker нет. Ограниченный параллелизм для обработки набора элементов даёт parallelMap.

При обычном запуске происходит следующее:

  1. Сохраняется текущий Log4j ThreadContext (MDC).

  2. Задача передаётся executor.

  3. В виртуальном потоке восстанавливается MDC, увеличивается async_worker_active_tasks, задача появляется в runningTasks().

  4. Создаётся транзакционный ServerContext указанного пользователя.

  5. После успешного Runnable контекст выполняет commit и закрывается.

  6. Ошибка журналируется, задача получает статус failure; успешная — success.

  7. Публикуются длительность и счётчик выполнения, число активных задач уменьшается.

Жизненный цикл задачи
Figure 1. Жизненный цикл задачи

Контекст вызывающего потока и JDBC-соединения в задачу не переносятся. Передавайте только неизменяемые значения: идентификаторы, строки и DTO. Не захватывайте ServerContext, Connection, service/repository instance, DynActionForm или изменяемое поле singleton-listener.

API

Метод Поведение Когда применять

submit(type, user, task)

Запускает задачу в новом транзакционном ServerContext; user по умолчанию — USER_BGCRM

Обычная фоновая операция без управления отменой

submitCancellable(type, user, inContext, task)

Возвращает Future; при inContext = false не создаёт ServerContext

Цикл или составная задача, которой нужен внешний контроль жизненного цикла

submitAfterCommit(type, user, task)

В транзакционном контексте ставит задачу в executor только после успешного commit; без него запускает сразу

Работа опирается на изменения текущей транзакции

submitWithLoading(message, type, user, task)

Показывает пользователю индикатор, запускает задачу после commit, при ошибке отправляет ShowErrorEvent, после выполнения или отката скрывает индикатор

Пользовательская операция с долгим внешним вызовом

submitWithProgress(message, type, user, task)

Как submitWithLoading, но задача получает AsyncProgress для уведомлений пользователю по ходу работы

Операции, которые сообщают промежуточный результат или открывают процесс

supply(type, user, task)

Выполняет Callable в транзакционном контексте и возвращает CompletableFuture с результатом или ошибкой

Нужен результат фоновой работы

parallelMap(type, user, items, maxParallel, mapper)

Обрабатывает элементы параллельно, не больше maxParallel одновременно, и возвращает результаты в исходном порядке; блокирует вызывающего

Массовый запрос к внешней системе или БД с ограничением нагрузки

submitWithRetry(type, user, attempts, backoff, task)

Повторяет задачу при исключении, каждая попытка в новом контексте, пауза удваивается

Идемпотентный вызов внешней системы с кратковременными сбоями

submitUnique(type, key, user, task)

Не запускает задачу, если задача с тем же type и key ещё в очереди или выполняется; возвращает false

Защита от двойного нажатия и повторных событий

runningTasks()

Список выполняющихся задач: номер, тип, пользователь, время старта

Диагностика зависаний и инцидентов

close()

Запрещает приём новых задач и ждёт завершения запущенных до 30 секунд

Вызывается ShutdownHook, не прикладным кодом

type — статическое имя класса операций, например catalog_export. Не включайте в него process ID, UUID, номер договора или логин: значение становится тегом метрики, и динамические части создают высокую кардинальность. Ключ дедупликации передаётся в submitUnique отдельным аргументом key и в метрики не попадает.

У Java-вызова параметр user можно опустить во всех методах, кроме submitWithProgress: для них сгенерированы перегрузки без него.

Базовый запуск

val processId = process.id

AsyncWorker.submit("catalog_export", user) {
    exportCatalog(processId)
}

Вызов возвращает управление после передачи задачи executor, а не после её завершения. Результат фоновой операции нельзя возвращать в исходный HTTP-response.

Если user не передан, задача выполняется как USER_BGCRM. Для аудита изменений от имени сотрудника и пользовательских уведомлений передавайте реального пользователя явно.

Запуск после commit

Задача, порождённая изменением БД, должна стартовать после подтверждения исходной транзакции:

AsyncWorker.submitAfterCommit("catalog_export", user) {
    exportCatalog(processId)
}

В транзакционном ServerContext метод регистрирует callback через invokeAfterCommit. Если commit не состоялся, задача не передаётся executor. Это предотвращает гонку, когда фоновый поток читает ещё не подтверждённые данные или выполняет внешний побочный эффект для откатившейся операции.

Не пишите вместо этого context.invokeAfterCommit) → AsyncWorker.submit(…​ вручную: submitAfterCommit делает то же самое и корректно работает без контекста.

Последовательность для commit и отката показана на схеме в разделе submitWithLoading.

Без текущего контекста и в transactional = false отложить запуск не к чему, поэтому задача передаётся executor сразу.

callback после commit существует только в памяти. Сбой JVM между commit БД и передачей задачи executor всё равно может потерять работу. Если потеря недопустима, сохраняйте запись задания в той же транзакции и выбирайте её устойчивым обработчиком.

Парный механизм ServerContext.invokeAfterRollback(callback) выполняет callback, если транзакция завершилась без commit: при rollback() или при close() без предшествующего commit(). Успешный commit снимает такие callback’и. Им пользуется submitWithLoading, чтобы убрать индикатор задачи, которая из-за отката не запустится.

Контекст и транзакция

Все методы, кроме submitCancellable(…​, inContext = false, …​), вызывают задачу внутри runWithContext(user, transactional = true). Соединения открываются лениво, а успешное завершение подтверждает все использованные соединения.

Если Runnable выбросил исключение, runWithContext не вызывает commit и закрывает контекст без явного rollback(). Исключение затем перехватывает сам AsyncWorker.

Для долгого внешнего ожидания не открывайте JDBC-соединение заранее. Предпочтительный порядок:

  1. выполнить внешний вызов;

  2. затем получить service/repository из контекста;

  3. коротко записать результат;

  4. вернуть управление, чтобы runWithContext выполнил commit.

Если до внешнего вызова требуется чтение из БД, а после него — отдельная запись, используйте submitCancellable(…​, inContext = false, …​) и создавайте короткие контексты withContext только вокруг фаз БД. Подробная схема приведена в разделе про внешние сервисы.

Ошибки и результат

AsyncWorker перехватывает любое исключение задачи, записывает его в лог и телеметрию со статусом failure и не передаёт потоку, который вызвал submit.

Ошибки уровня Error (NoClassDefFoundError после перекомпиляции динамического кода, StackOverflowError, OutOfMemoryError) тоже журналируются и учитываются в телеметрии, после чего пробрасываются дальше. У submitCancellable и submitWithRetry такой Error придёт в Future.get() как ExecutionException.

У submitCancellable возвращаемый Future предназначен прежде всего для отмены. Обычное исключение задачи перехватывается внутри worker, поэтому Future.get() после такого сбоя не получает ExecutionException. Если вызывающему нужен результат или ошибка, используйте supply. Бизнес-результат долгой операции сохраняйте в состоянии операции и/или отправляйте пользователю через ClientEvents.

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

Результат задачи: supply

val balance: CompletableFuture<BigDecimal> = AsyncWorker.supply("balance_load", user) {
    billing.getBalance(contractId)
}
Результат задачи через supply
Figure 2. Результат задачи через supply

Future завершается после commit контекста задачи: значение, записанное в БД внутри задачи, к этому моменту уже подтверждено. Исключение задачи или commit завершает future этим исключением, и оно же учитывается в телеметрии как failure.

cancel() у возвращённого CompletableFuture не прерывает уже запущенную задачу. Не ждите результат в HTTP-запросе с открытой транзакцией дольше, чем ждали бы синхронный вызов: соединение запроса остаётся занятым всё время ожидания.

Параллельная обработка: parallelMap

List<ContractInfo> infos = AsyncWorker.parallelMap(
        "contract_info_load", USER_BGCRM, contractIds, 8,
        contractId -> billing.getContractInfo(contractId));
Ограничение параллелизма в parallelMap
Figure 3. Ограничение параллелизма в parallelMap

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

Метод блокирует вызывающий поток до обработки всех элементов и возвращает результаты в порядке items. Ошибка одного элемента не останавливает остальные. Если ошибки были, после обработки всех элементов выбрасывается BGException: первая ошибка в cause, остальные в suppressed. По каждому элементу публикуется отдельное событие телеметрии с одним и тем же type.

Не вызывайте parallelMap для сотен тысяч элементов из HTTP-запроса: запрос будет ждать всю обработку. Для таких объёмов используйте модуль Task.

Повтор: submitWithRetry

AsyncWorker.submitWithRetry("camera_registration", user, attempts = 3, backoff = Duration.ofSeconds(2)) {
    cameraService.register(processId)
}
Повтор с удвоением паузы
Figure 4. Повтор с удвоением паузы

При исключении задача повторяется, пока не исчерпано attempts попыток. Первая пауза равна backoff, каждая следующая вдвое длиннее: в примере 2 и 4 секунды. Каждая попытка выполняется в новом контексте, поэтому изменения неудачной попытки в БД откатываются. Промежуточные сбои пишутся в лог с уровнем WARN, в телеметрию попадает одно событие на всю задачу с итоговым статусом.

Повторяйте только идемпотентные операции. Если внешняя система могла выполнить команду, но ответ потерялся, повтор создаст дубль. Подробнее — в разделе про внешние сервисы.

Дедупликация: submitUnique

boolean started = AsyncWorker.submitUnique("contract_sync", String.valueOf(processId), user,
        () -> syncContract(processId));
if (!started) {
    ClientEvents.INSTANCE.showInfo(user.getId(), "Синхронизация уже выполняется");
}
Дедупликация по ключу
Figure 5. Дедупликация по ключу

Пока задача с тем же type и key стоит в очереди или выполняется, повторные вызовы возвращают false и задачу не запускают. После завершения, успешного или нет, ключ освобождается. Ключ проверяется только на текущем инстансе: запрос на другой инстанс ERP запустит задачу параллельно.

Отмена

submitCancellable позволяет вызвать Future.cancel(…​), но отмена остаётся кооперативной:

  • cancel(false) не останавливает уже начавшуюся задачу;

  • cancel(true) прерывает виртуальный поток, только если вызываемый код реагирует на interrupt;

  • AsyncWorker не компенсирует уже выполненный внешний или локальный побочный эффект.

Прерывание виртуального потока может немедленно оборвать JDBC или сетевую операцию. Код cleanup в том же прерванном потоке тоже может потерять возможность обратиться к БД. Не используйте cancel(true) для операции, которая обязана после отмены записать состояние или освободить бизнес-захват через БД; проектируйте отдельный подтверждаемый протокол отмены.

submitWithLoading

Метод требует пользователя с положительным id. Индикатор отправляется:

  • в JSON-response текущего HTTP-запроса, если его form принадлежит тому же пользователю;

  • через Redis-канал ClientEvents, если подходящего response нет.

Сама задача стартует после commit исходного транзакционного контекста. После старта HideLoadingEvent отправляется в finally, а исключение или Error дополнительно создаёт ShowErrorEvent и статус failure в телеметрии.

Если исходная транзакция откатилась или контекст закрылся без commit, задача не запускается, а HideLoadingEvent уходит через Redis-канал. Индикатор остаётся только при падении JVM между показом и завершением задачи.

Запуск после commit и откат транзакции
Figure 6. Запуск после commit и откат транзакции

Полный пример и канал доставки: Клиентские события.

submitWithProgress

Вариант submitWithLoading, в котором задача получает объект AsyncProgress. Через него можно по ходу работы отправить пользователю info(…​), error(…​) и openProcess(processId); вызовы возвращают тот же объект и объединяются в цепочку. Получателем всегда будет пользователь, запустивший операцию, так что id не нужно захватывать отдельно.

AsyncWorker.submitWithProgress("Загрузка адресов", "fias_loader", user, progress -> {
    int count = loader.load(processId);
    progress.info("Загружено адресов: " + count).openProcess(processId);
});

Индикатор скрывает сам AsyncWorker, progress.error(…​) не прерывает задачу. Правила для user, commit и отката — те же, что у submitWithLoading. У метода нет перегрузок без type и user.

Телеметрия, диагностика и завершение JVM

Метрика Содержание

async_worker_tasks_total

Завершённые задачи: теги event=<type>, status=success|failure

async_worker_duration_milliseconds

Полная длительность задачи с теми же тегами

async_worker_active_tasks

Задачи, которые уже начали выполняться; ожидающие старта не учитываются

runningTasks() возвращает снимок задач, которые выполняются в данный момент: порядковый номер, type, id пользователя и время старта. По нему видно, какая задача висит дольше ожидаемого.

close() вызывает ExecutorService.shutdown() и ждёт завершения уже запущенных задач до 30 секунд. Если за это время они не завершились, в лог пишется предупреждение с их типами, а JVM продолжает остановку. Задачи не сохраняются, критичный результат должен восстанавливаться из БД после следующего запуска.

Ограничения

AsyncWorker сам по себе не предоставляет:

  • постоянную очередь и повтор после рестарта;

  • общее ограничение параллелизма и backpressure (только в рамках одного вызова parallelMap);

  • timeout задачи;

  • дедупликацию и координацию между ERP-инстансами;

  • сохранение бизнес-статуса или результата.

Эти свойства должны обеспечиваться вызывающей подсистемой и штатным клиентом внешнего API.