Привет, Хабр! Наше подразделение занимается автоматизацией бизнес-процессов обслуживания клиентов в офисах и отделениях банка. Такие бизнес-процессы специфичны для конкретной организации, связаны с соблюдением внутренних регламентов и нормативных документов банка, должны полностью соответствовать часто меняющимся требованиям законодательства и всем инструкциям и правилам ЦБ РФ. Так, например, за процедурой открытия вклада стоит множество взаимодействий различных классов систем и компонентов, начиная от сбора и хранения согласий клиента на обработку персональных данных, интеграций с госсервисами и системами сбора сведений и заканчивая формированием приходно-кассовых ордеров. 

Подобные процессы мы реализуем с помощью движка Camunda, который позволяет прототипировать и визуализировать сложную последовательность шагов и моделировать flow совместно с бизнес-заказчиком, а также обладает множеством инструментов для реализации интеграций, оркестрации, мониторинга, сбора бизнес-метрик и тестирования. + быстро и относительно незатратно менять процессы (можно привести пример с запретом перевода валюты в другие банки). Вдобавок Camunda является open-source решением, что крайне важно в текущих реалиях и т.д. 

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

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

Архитектура 

Мы провели сравнение Camunda 8 и Camunda 7 по нашим внутренним критериям, проанализировали варианты развёртывания Camunda 7 (https://docs.camunda.org/manual/7.19/introduction/architecture/) и остановились на кластерной модели Camunda 7 с рекомендованной разработчиками Camunda архитектурой развёртывания c использованием удаленного движка, а точнее Camunda Run в качестве workflow-движка. 

Далее мы развернули отдельный инстанс СУБД Postgres только под задачи Camunda, несколько экземпляров Camunda Run (воркеры) в кластере за балансировщиком для исполнения процессов, а также отдельный независимый экземпляр Camunda Run исключительно для задач администрирования и мониторинга. 

Настройки Camunda Run для администрирования и мониторинга процессов: 

camunda.bpm: 
  job-execution: 
    enabled: false  

Настройки узлов Camunda Run (воркеров) за балансировщиком 

camunda.bpm: 
  job-execution: 
    max-wait: 10000 
    backoff-time-in-millis: 50 
    max-backoff: 250 
    backoff-decrease-threshold: 100 
    wait-increase-factor: 2 

 
Из коробки Camunda не умеет работать с Kafka, но большая часть наших процессов и внутренних систем базируются на event-driven паттерне, поэтому после первого этапа выбора архитектуры и развёртывания системы нам было необходимо подружить Camunda с Kafka: 

• для возможности автоматического запуска определённых бизнес-процессов по мере появления потребности 

• для запроса действий в других системах в рамках исполнения различных бизнес-задач 

• для обеспечения интеграционных взаимодействий  

Укрупнённо план выглядел следующим образом: 

Где...  

Микросервисы активации процессов:

• являются держателями спроектированных bpmn-диаграмм 

• могут вести учёт и версионирование диаграмм 

• взаимодействуют с Kafka и умеют соотносить нужный процесс с тем или иным событием 

• обеспечивают идемпотентность на случай дублей или сбоев 

Микросервисы обработки внешних задач :

• используются вместо делегатов 

• обращаются к Camunda с заданной частотой для получения задач 

• могут исполнять сложную логику (например, сравнения данных из различных систем) 

• могут запускать другие bpmn-процессы (через Kafka и взаимодействия с микросервисами активации процессов) 

• позволяют извлечь необходимые процессу данные из других систем 

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

Здесь стоит сказать, что для взаимодействия с Kafka мы используем модель, предложенную Крисом Ричардсоном, 

Асинхронное API наших сервисов состоит из каналов и соответствующих типов сообщений, таких как: 

• входящие команды, 

• ответы на командные запросы и 

• события 

Схема асинхронного API сервиса из книги Ричардсона приведена ниже: 

Таким образом, возвращаясь к bpmn-схеме, кубик  «Отправка запроса открытия вклада» должен быть реализован отправкой командного события в топик Kafka. 

Внешняя задача для данного шага была реализована с использованием базового класса обработки задач, наследующего интерфейс библиотеки Camunda  

org.camunda.bpm.client.task.ExternalTaskHandler 

с преобразованием атрибутов объекта ExternalTask в расширенный под наши потребности DTO: 

/**  
* Обобщенное выполнение задачи.
*  
* @param externalTask        внешняя задача  
* @param externalTaskService сервис внешних задач  
*/ 
@Override 
public void execute(@NonNull final ExternalTask externalTask,
              @NonNull final ExternalTaskService externalTaskService) {
  final TaskBO task = ExternalTaskToTaskBOMapper.EXTERNAL_TASK_TO_TASK_BO_MAPPER.map(externalTask); 

Для сквозной и межсистемной трассировки мы передаём в хедеры сообщения в Kafka корреляционный ключ инстанса работающего бизнес-процесса (correlationId), и этот же идентификатор вернётся нам в хедерах reply-события для связки ответа микросервиса открытия вклада с конкретным исполняющимся процессом.  

/**  
* Отправка сообщения в топик   
*  
* @param task    DTO-представление внешней задачи  
* @param message DTO для отправки в топик Kafka  
*/ 
public void sendCommand(@NonNull final TaskBO task,                         
              @NonNull final Object message) {     
  kafkaProducer.sendCommand(task.getBusinessKey(), message, task.getHeaders()); 
} 

Кубик получения результата открытия депозита из топика Kafka также реализован внешней задачей. 

Мы реализовали подписку на заданный reply-топик микросервиса открытия депозитов и ожидали в нём появления ответного события по первоначальному командному запросу. 

String processCorrelationId =  externalTask.getVariable(CORRELATION_ID); 
correlationIds.add(processCorrelationId);
executor.execute(() -> {
     try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>
     (defaultConsumerProperties)) {
              consumer.subscribe(Collections.singletonList(externalTask.getVariable(KAFKA_TOPIC)));
     	      ...
     }
     ... 

Пока инстанс микросервиса обработки внешних задач был один – всё работало отлично, каждое входящее событие из reply-топика Kafka фильтровалось consumer-ом по бизнес-ключу и хедерам, и в случае совпадения значения correlationId в хедере с любым элементом внутреннего списка микросервиса correlationIds, задача успешно завершалась. Однако, как только мы пытались масштабироваться и увеличить количество инстансов микросервисов обработки задач – часть reply-событий для одного инстанса-подписчика попадали в обработку другого инстанса-подписчика, correlationId входящего события отсутствовал во внутреннем списке инстанса correlationIds, и исполнение внешних задач в процессе отваливалось по таймауту. 

Для исправления данной ситуации потребовалось переопределить свойства подписчика и помещать его в уникальную группу для того чтобы каждый инстанс микросервиса исполнения внешних задач получал каждое событие из нашего reply-топика Kafka: 

Properties consumerProperties = new Properties(defConsumerProperties); 
consumerProperties.putAll(defConsumerProperties); 
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_TASK_GROUP_ID + UUID.randomUUID()); 

И далее применить данные свойства к consumer: 

executor.execute(() -> {
     try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties)) 
{
          String kafkaTopic = externalTask.getVariable(KAFKA_TOPIC);
          consumer.subscribe(Collections.singletonList(kafkaTopic)); 

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

Собственно, данных артефактов и манипуляций было достаточно для реализации и исполнения приближенного к реальному бизнес-процесса открытия вклада и проведения итерации нагрузочного тестирования (для того чтобы убедиться в жизнеспособности схемы под нагрузкой на проде). 

Первая итерация нагрузочного тестирования включала в себя один инстанс воркера (Camunda Run), 1 инстанс сервиса активации процесса и один инстанс сервиса обработки внешних задач и стартовые показатели ресурсов:

Настройки деплоймента для микросервиса обработки внешних задач:

resources: 
  limits: 
    cpu: '1' 
    memory: 512Mi 
  requests: 
    cpu: '1' 
    memory: 512Mi 

Настройки деплоймента для микросервиса обработки активации процессов:

resources: 
  limits: 
    cpu: '1' 
    memory: 512Mi 
  requests: 
    cpu: '1' 
    memory: 512Mi  

Настройки деплоймента воркера:    

resources: 
  limits: 
    cpu: '4' 
    memory: 2G 
  requests: 
    cpu: '1' 
    memory: 512Mi 

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