AsyncWorker

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

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

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

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

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

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

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

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

  3. В виртуальном потоке восстанавливается MDC и увеличивается async_worker_active_tasks.

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

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

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

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

Контекст вызывающего потока и 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, после выполнения скрывает индикатор

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

close()

Запрещает приём новых задач вызовом shutdown()

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

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

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

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. Это предотвращает гонку, когда фоновый поток читает ещё не подтверждённые данные или выполняет внешний побочный эффект для откатившейся операции.

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

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

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

submit, submitAfterCommit и submitWithLoading вызывают задачу внутри runWithContext(user, transactional = true). Соединения открываются лениво, а успешное завершение подтверждает все использованные соединения.

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

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

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

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

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

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

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

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

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

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

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

Отмена

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, а исключение дополнительно создаёт ShowErrorEvent и статус failure в телеметрии.

Гарантия скрытия относится к уже запущенной задаче. Если исходная транзакция не дошла до commit, callback не запускается. В HTTP-сценарии response с индикатором обычно тоже не становится успешным, но при внеполосной отправке индикатор может оставаться до вытеснения/истечения TTL; этот случай нужно учитывать при ручном использовании метода вне request-response.

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

Телеметрия и завершение JVM

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

async_worker_tasks_total

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

async_worker_duration_milliseconds

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

async_worker_active_tasks

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

close() вызывает ExecutorService.shutdown(), но не ждёт завершения задач и не сохраняет их. На остановке приложения нельзя рассчитывать на завершение фоновой операции. Критичный результат должен восстанавливаться из БД после следующего запуска.

Ограничения

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

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

  • ограничение параллелизма и backpressure;

  • timeout задачи;

  • автоматический retry;

  • дедупликацию и идемпотентность;

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

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

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