Неблокирующий вызов внешнего сервиса в процессе

Довольно часто из процесса приходится обращаться по API к различным внешним сервисам. По сути, это стандартный сценарий оркестрации систем или даже микросервисов (пардон за модное слово). На BPMN диаграмме это выглядит просто и логично – постучались куда-то по API, получили ответ и перешли к следующей задаче. Для моделей аналитического уровня это нормально.

Синхронное и асинхронное выполнение

BPM-движки поддерживают два вида выполнения задач – синхронное и асинхронное. На мой взгляд, названия крайне неудачные, только напускают туману и создают путаницу. На этом сломался даже искусственный интеллект. Я спрашивал ChatGPT и Perplexity, как они понимают синхронное и асинхронное выполнение задач в BPMN и оба с первого раза лажанулись.  

ChatGPT: Асинхронное выполнение означает, что процесс может продолжить выполнение, даже если текущая задача еще не завершена. Это используется в случаях, когда:

• Задача выполняется долго (например, сложная обработка данных).

• Не требуется немедленного завершения перед переходом к следующему шагу.

• Требуется повысить отказоустойчивость (например, если сервис временно недоступен).

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

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

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

На самом деле это все про границы транзакций, а не про какую-то реальную асинхронность. В контексте BPMN асинхронное выполнение задач связано с тем, как процесс сохраняет свое состояние и управляет выполнением шагов.

Когда вы используете асинхронные задачи (например, с атрибутами asyncBeforeили asyncAfter в Camunda или просто asyncв Flowable), процесс явно разделяется на независимые транзакции. Это означает, что состояние процесса фиксируется в точке разделения (то есть записывается в базу данных), и дальнейшее выполнение передается в очередь задач (Job Queue). После этого задача обрабатывается движком процесса (например, Camunda или Flowable) в отдельной транзакции.

Таким образом, асинхронность в BPMN — это способ управления транзакциями и обработкой задач через очереди, а не параллельное выполнение кода или потоков.
 

Синхронные задачи

Сервисные задачи по умолчанию создаются как синхронные. Определенная логика в этом есть – так движок работает быстрее, потому что не надо сохранять состояние процесса в БД после каждой задачи. Все выполняется в рамках одной транзакции, пока на пути не встретится user task или другой элемент, вызывающий состояние ожидания. Для простых задач, в которых, например, изменяется состояние процессной переменной, производятся нехитрые вычисления или запись в лог, это прекрасно работает. А в случаях, когда результат не гарантирован и может произойти сбой, синхронное выполнение крайне нежелательно. Потому что произойдет откат транзакции и все результат предыдущих задач будет аннулирован и процесс выкинет ошибку. Оно вам надо?

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

Асинхронные задачи и Fail Retry

Теперь рассмотрим асинхронные задачи. Как вы уже поняли из нашего диспута с ИИ, асинхронность в BPMN не подразумевает независимого выполнения. Токен все равно не двинется дальше до тех пор, пока задача не завершится. Вся разница в том, что теперь за выполнение задачи отвечает не сам BPM-движок, а специальный компонент Job Executor . То есть, все асинхронные задачи из всех процессов скидываются в общую очередь и этот самый Job Executor в порядке той самой очереди их выполняет. Если задача завершается успешно, Job Executor уведомляет об этом процесс и токен двигается дальше.

То есть, мы искусственно создали новую границу транзакции и теперь результат предыдущих задач не будет отменен, если что-то пойдет не так:

Окей, а что произойдет, если наша задача прервалась по таймауту или выбросила exception? В терминологии BPMN это будет так называемая «техническая ошибка» и если установлен параметр fail retry (счетчик повторных попыток), то Job Executor попытается выполнить ее еще раз. А потом еще и так пока счетчик не обнулится. Когда это произойдет, Camunda создаст инцидент, а Flowable пометит эту задачу как Failed. Дальше админ может попытаться как-то исправить эту ситуацию вручную.

Довольно примитивное решение, надо сказать! Ну а что вы хотели, это было придумано двадцать лет назад, тогда казалось нормально. Никто никуда не спешил, процессы были преимущественно с участием людей, а человек это всегда тормоз для процесса. Поэтому до сих пор значение параметра Fail Retry в Camunda и Flowable 3R5M, что означает повторить три раза через пять минут. Для полностью автоматизированных процессов, заточенных на оркестрацию микросервисов, пауза между попытками в пять минут выглядит как вечность. В борьбе за эффективность счет идет на миллисекунды, а тут такое…

Почему бы не поставить fail retry например, cто раз через сто миллисекунд? – Не прокатит. Этот механизм работает так же, как Timer event. То есть, в БД создается специальная запись и уже знакомый нам Job Executor периодически проверяет, пора этому таймеру щелкнуть или еще нет. Из практики, минимальное значение должно быть порядка 20 секунд, за меньшее время это все не срабатывает. Конечно, зависит от железа, но по-любому это небыстро.

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

То есть, пора признать, что это устаревший механизм, который не стоит применять в решениях, ориентированных на полностью автоматизированные процессы с высокой интенсивностью. – И что же делать? – Можно посмотреть в сторону паттерна Circuit Breaker, но это другая история, об этом в следующий раз.
 

Прикрепленные (граничные) события

В BPMN есть еще один механизм, который можно задействовать в нашем сценарии с ненадежным внешним сервисом. Речь пойдет о прикрепленных или граничных событиях (boundary events). С их помощью можно реализовать сценарий, как с fail retry, только более гибкий. Например, по таймауту вы может либо точно также уйти на повторную попытку, либо эскалировать эту задачу на какого-то сотрудника, чтобы он выполнил ее вручную.

Допустим, ваш процесс автоматически обращается к некоему сервису проверки контрагентов, а на той стороне поменялось API и все упало. Вместо того, чтобы подвешивать инциденты, можно делегировать эту задачу человеку, который может просто позвонить вашему партнеру. Аналогично и с обработкой ошибок. Если сервис вернул какой-то нехороший код, можно смоделировать любую логику, как на это реагировать. Это гораздо гибче, чем fail retry и визуально понятнее. Хотя схема становится сложнее. Но жизнь вообще штука сложная.

Однако, вызов сервиса при этом все равно остается блокирующим – процесс ждет завершения задачи. И при этом у нас висит открытая транзакция.

–  Что произойдет, если сервер упадет при открытой транзакции?

Чего-то совсем фатального не случится. Произойдет откат транзакции, потому что сам движок использует двухфазный коммит, это нормальное поведение. Однако, если транзакция не была зафиксирована, система BPM-движок при перезапуске обнаружит незавершенную задачу в таблице ACT_RU_JOB и обработает ее снова. То есть, задача будет выполнена повторно. А если она не идемпотентна, это может привести к дублированию данных. Это справедливо как для синхронного, так и для асинхронного режима.

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

Event-based gateway

Event-based gateway — это элемент BPMN, который используется для принятия решений на основе событий. В отличие от других шлюзов, таких как exclusive gateway, которые принимают решения на основе данных, event-based gateway ожидает наступления одного из нескольких возможных событий и выбирает путь в процессе в зависимости от того, какое событие произошло первым.

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

На бумаге выглядит красиво! Но так не работает, если делать вызов API из обычного JavaDelegateили Spring Bean посредством старого доброго RestTemplate. (Конечно, сам RestTemplateэто синхронный HTTP-клиент, который блокирует поток, пока выполняется запрос. Но этот способ проще и привычнее для многих, нежели
WebClient, поэтому будем использовать его.)

Нам всего-то надо позволить завершиться сервисной задаче, чтобы процесс мог двинуться дальше и перейти в состояние ожидания с event based gateway. Но как же быть? – А давайте отправлять HTTP-запрос не из самого сервис-таска, а из другого места. Пусть сервис-таск только инициирует это действие посредством события приложения (application event). Тогда все получится: наш сервис-таск по-быстрому выстрелит событие и его дело сделано. Затем событие подхватит листенер, но уже вне контекста процесса. Тот, в свою очередь, выполнит HTTP-запрос при помощи RestTemplate, дождется ответа и отправит в процесс то или иное сообщение.

Вот так это выглядит на временной диаграмме. Мы развязали исполнение BPMN-процесса и обращение к внешнему сервису. Процесс продолжится по тому или иному пути в зависимости от результата запроса, ОК или ошибка.

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

Реализуем процесс на Jmix BPM с движком Flowable

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

Open-source geocoding with OpenStreetMap data

Nominatim (от латинского, «по имени») — это инструмент для поиска данных OSM по названию и адресу, а также для генерации синтетических адресов точек OSM (обратное геокодирование). Он также имеет ограниченную возможность поиска объектов по их типу (пабы, отели, церкви и т. д.) Имеет API, которое мы и будем использовать.

В центре диаграммы вы видите сервисную задачу Send request to Nominatim. Как мы обсуждали выше, в самой задаче API-вызов не производится, из нее только отправляется application event, который ловит наш листенер. На BPMN-схеме его нет, но он выполняет самую важную работу, непосредственно геокодинг. После сервисной задачи стоит event based gateway, настроенный на три варианта продолжения процесса:

• Нормальный, когда получен положительный ответ, то есть по адресу нашлись реальные координаты.

• Ошибка, когда адрес оказался фейковым и сервис вернул null. В этом случае наш процесс выбрасывает BPMN-ошибку с определенным кодом, чтобы ее можно было обработать в процессе верхнего уровня.

• Выход по таймауту. Естественно, внешний сервис может быть недоступен.

Необязательно, чтобы сервис сам был в ауте, может просто вы вне зоны доступа и приложение не видит интернет. Но процесс при этом не должен тупо висеть, эта ситуация должна корректно обрабатываться. Поэтому мы еще смоделируем fail retry – дадим три попытки все-таки получить ответ. При каждой новой попытке счетчик увеличивается и после трех раз выбрасываем BPMN-ошибку, но с другим кодом.

Чем это отличается от стандартного подхода:

• Полностью устранены риски, связанные с долгими транзакциями в процессе.

• Все исключительные ситуации на виду и их можно дорабатывать. Например, после трех попыток не просто бросать ошибку, а эскалировать задачу на человека. Или можете придумать что-то свое.

Реализация

В сервисной задаче Send request to Nominatim вызывается бин AddressVerificator :

@Component(value = "ord_AddressVerificator")
public class AddressVerificator {

    private static final Logger log = LoggerFactory.getLogger(AddressVerificator.class);

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    public void verify(Order order, DelegateExecution execution) {
        String processInstanceId = execution.getProcessInstanceId();
        if (order.getAddress() != null) {
            applicationEventPublisher.publishEvent(new RequestSendEvent(this, order, processInstanceId));
        } else {
            log.error("Order # {}: Address is NULL", order.getNumber());
        }
    }
}

На самом деле, из метода verify этого бина только отправляется сообщение RequestSendEvent при помощи ApplicationEventPublisher, больше никакой логики в нем нет.

Это сообщение ловит листенер onRequestSendEvent:

@Async
@EventListener
public void onRequestSendEvent(RequestSendEvent event) {
    Order order = event.getOrder();
    String processInstanceId = event.getProcessInstanceId();
    Point point = geoCodingService.verifyAddress(order.getAddress());

    if (point != null) {
        orderService.setLocation(order, point);
        sendMessageToProcess("Address OK", processInstanceId);
        log.info("Order # {}, Address verified: {}", order.getNumber(), order.getAddress());
    } else {
        sendMessageToProcess("Fake address", processInstanceId);
        log.info("Order # {}, Invalid address: {}", order.getNumber(), order.getAddress());
    }
}

private void sendMessageToProcess(String messageName, String processInstanceId) {
    ProcessInstance processInstance = runtimeService.createProcessInstanceQuery()
            .processInstanceId(processInstanceId)
            .singleResult();
    if (processInstance == null) return;
    Execution execution = runtimeService.createExecutionQuery()
            .messageEventSubscriptionName(messageName)
            .parentId(processInstance.getId())
            .singleResult();
    if (execution == null) return;
    runtimeService.messageEventReceivedAsync(messageName, execution.getId());
}

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

Аннотация @Asyncв Spring используется для выполнения методов в асинхронном режиме. Это означает, что метод, помеченный этой аннотацией, будет выполняться в отдельном потоке, не блокируя основной поток выполнения программы. Таким образом, основной поток может продолжать выполнять другие задачи, не дожидаясь завершения асинхронного метода.

Далее, выполняется вызов метода verifyAddress из бина GeoCodingService, где и происходит обращение к сервису Nominatim. Метод возвращает точку на карте – объект Point. В зависимости от результата, в процесс отсылается сообщение “Address OK” или “Fake address”. И тогда срабатывает event based gateway.

Сервис геокодирования выполняет совершенно типовой HTTP-запрос при помощи RestTemplate:

@Service
public class GeoCodingService {
    private final RestTemplate restTemplate;

    public GeoCodingService(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    public Point verifyAddress(String address) {
        String NOMINATIM_URL = "https://nominatim.openstreetmap.org/search?q={address}&format=json&polygon_kml=1&addressdetails=1";

        ResponseEntity<NominatimResponse[]> response = restTemplate.getForEntity(NOMINATIM_URL,
                NominatimResponse[].class,
                address);
        if (response.getStatusCode().is2xxSuccessful() && response.getBody() != null) {
            NominatimResponse[] results = response.getBody();
            if (results.length > 0) {
                double latitude = results[0].latitude();
                double longitude = results[0].longitude();
                GeometryFactory geometryFactory = new GeometryFactory();
                return geometryFactory.createPoint(new Coordinate(longitude, latitude));
            }
        }
        return null;
    }
}

Если сервис не ответил за определенное время, срабатывает таймер и выполняется повторная попытка. Чтобы не получить бесконечный цикл, используем счетчик повторов. В принципе, после исчерпания попыток можно эскалировать эту задачу на сотрудника-человека, который выполнит проверку вручную.

Подведем итоги

Проектируя бизнес-процесс, нужно обращать внимание на длительность выполнения сервисных задач и возможные исключительные ситуации и корректно их обрабатывать. К механизму fail retry следует относится с осторожностью. Он был придуман очень давно, сейчас есть более продвинутые паттерны для подобных ситуаций.

Комбинируя события приложения и события BPMN, можно развязать выполнение запроса к внешнему сервису и завершение задачи, чтобы использовать event based gateway. Это уменьшает риски из-за длительных транзакций.

Подписывайтесь на Telegram каналы: 

Jmix.ru — платформа быстрой разработки B2B и B2G веб-приложений на Java.

BPM Developers — про бизнес процессы: новости, гайды,  полезная информация и юмор.