🥇Інтеграція в стилі BPM

Привет, Хабр!

Наша компанія спеціалізується на розробці програмних рішень класу ERP, у складі яких левову частку займають транзакційні системи з величезним обсягом бізнес-логіки та документообігом а-ля СЕД. Сучасні версії наших продуктів базуються на технологіях JavaEE, але ми також активно експериментуємо із мікросервісами. Одне з проблемних місць таких рішень - інтеграція різних підсистем, що відносяться до суміжних доменів. Завдання інтеграції завжди доставляли нам величезний головний біль, незалежно від застосовуваних нами архітектурних стилів, технологічних стеків та фреймворків, проте останнім часом у вирішенні таких завдань намітився прогрес.

У статті, що пропонується до вашої уваги, я розповім про наявні у НВО «Криста» досвід та архітектурні дослідження в зазначеній області. Також ми розглянемо приклад простого вирішення інтеграційного завдання з погляду прикладного розробника і з'ясуємо, що ховається за цією простотою.

Дисклеймер

Описані у статті архітектурні та технічні рішення пропонуються мною з урахуванням особистого досвіду у тих конкретних завдань. Ці рішення не претендують на універсальність і можуть виявитися не оптимальними за інших умов використання.

До чого тут BPM?

Для відповіді на це питання потрібно трохи заглибитись у специфіку прикладних завдань наших рішень. Основна частина бізнес-логіки в нашій типовій транзакційній системі – це введення даних у БД через інтерфейси користувача, ручна та автоматизована перевірка цих даних, проведення їх за деяким workflow, публікація в іншу систему/аналітичну базу/архів, формування звітів. Таким чином, ключовою функцією системи для замовників є автоматизація внутрішніх бізнес-процесів.

Для зручності ми використовуємо у спілкуванні термін «документ» як абстракцію набору даних, об'єднаних спільним ключем, до якого можна «прив'язати» певний workflow.
Але як бути з інтеграційною логікою? Адже завдання інтеграції породжується архітектурою системи, яка «розпиляна» на частини НЕ на вимогу замовника, а під впливом інших чинників:

під дією закону Конвею;
внаслідок повторного використання підсистем, раніше розроблених для інших продуктів;
за рішенням архітектора, виходячи з дисфункційних вимог.

Існує велика спокуса відокремити інтеграційну логіку від бізнес-логіки основного workflow, щоб не забруднювати бізнес-логіку інтеграційними артефактами та позбавити прикладного розробника необхідності вникати особливо архітектурного ландшафту системи. Такий підхід має ряд переваг, проте практика показує його неефективність:

вирішення інтеграційних завдань зазвичай скочується до найпростіших варіантів у вигляді синхронних викликів через обмеженість точок розширення в реалізації основного workflow (про недоліки синхронної інтеграції – трохи нижче);
інтеграційні артефакти все одно проникають в основну бізнес-логіку, коли потрібний зворотний зв'язок з іншої підсистеми;
прикладний розробник ігнорує інтеграцію та може легко її зламати, змінивши workflow;
система перестає бути єдиним цілим з погляду користувача, стають помітні «шви» між підсистемами, з'являються надмірні користувацькі операції, що ініціюють передачу даних з однієї підсистеми в іншу.

Інший підхід – розгляд інтеграційних взаємодій як невід'ємної частини основної бізнес-логіки та workflow. Щоб вимоги до кваліфікації прикладних розробників не злетіли до небес, створення нових інтеграційних взаємодій має виконуватися легко і невимушено, з мінімальними можливостями вибору способу рішення. Це зробити складніше, ніж здається: інструмент повинен бути достатньо потужним, щоб забезпечити користувачеві необхідну кількість варіантів його застосування і при цьому не дозволити «вистрілити собі в ногу». Існує безліч питань, на які повинен відповісти інженер у контексті інтеграційних завдань, але про які не повинен замислюватися прикладний розробник у своїй повсякденній роботі: межі транзакцій, консистентність, атомарність, безпека, масштабування, розподіл навантажень та ресурсів, роутинг, маршалінг, розповсюдження та перемикання контекстів тощо. Потрібно запропонувати прикладним розробникам досить прості шаблони рішень, у яких заховані відповіді всі подібні питання. Ці шаблони мають бути досить безпечними: бізнес-логіка змінюється дуже часто, що підвищує ризики внесення помилок, ціна помилок має залишатися на досить низькому рівні.

Але все-таки до чого тут BPM? Є безліч варіантів реалізації workflow…
Справді, у наших рішеннях дуже популярна інша реалізація бізнес-процесів – через декларативне завдання діаграми переходів станів та підключення обробників із бізнес-логікою на переходи. При цьому стан, що визначає поточне положення "документа" в бізнес-процесі, є атрибутом "документа".

Так виглядає процес на старті проекту

Популярність такої реалізації обумовлена відносною простотою та швидкістю створення лінійних бізнес-процесів. Однак у міру постійного ускладнення програмних систем автоматизована частина бізнес-процесу розростається та ускладнюється. З'являється необхідність декомпозиції, повторному використанні частин процесів, і навіть у розгалуженні процесів, щоб кожна гілка виконувалася паралельно. У таких умовах інструмент стає незручним, а діаграма переходів станів втрачає інформативність (інтеграційні взаємодії взагалі не позначаються на діаграмі).

Такий процес відбувається через кілька ітерацій уточнення вимог

Виходом із цієї ситуації стала інтеграція двигуна jBPM деякі продукти з найбільш складними бізнес-процесами. У короткостроковій перспективі це рішення мало певний успіх: з'явилася можливість реалізації складних бізнес-процесів із збереженням достатньо інформативної та актуальної діаграми у нотації. BPMN2.

Невелика частина складного бізнес-процесу

У довгостроковій перспективі рішення не виправдало очікувань: висока трудомісткість створення бізнес-процесів через візуальні інструменти не дозволила досягти прийнятних показників продуктивності, а сам інструмент став одним із найнелюбніших серед розробників. До внутрішнього пристрою двигуна теж були претензії, які призвели до появи безлічі «латок» та «милиць».

Головним позитивним моментом застосування jBPM стало усвідомлення користі та шкоди від наявності власного персистентного стану у екземпляра бізнес-процесу. Також ми побачили можливість застосування процесного підходу для реалізації складних протоколів інтеграції між різними програмами із застосуванням асинхронних взаємодій через сигнали та повідомлення. Наявність персистентного стану відіграє у цьому найважливішу роль.

З сказаного можна дійти невтішного висновку: процесний підхід у стилі BPM дозволяє нам вирішувати широкий спектр завдань з автоматизації бізнес-процесів, що постійно ускладнюються, гармонійно вписувати в ці процеси інтеграційні активності та зберігати можливість візуального відображення реалізованого процесу у відповідній для цього нотації.

Недоліки синхронних викликів як інтеграційного патерну

Під синхронною інтеграцією розуміється найпростіший блокуючий виклик. Одна підсистема виступає серверною стороною і виставляє API з необхідним способом. Інша підсистема виступає клієнтською стороною і в потрібний момент здійснює виклик з очікуванням результату. Залежно від архітектури системи клієнтська та серверна сторони можуть розміщуватися або в одному додатку та процесі, або в різних. У другому випадку потрібно застосувати деяку реалізацію RPC та забезпечити маршалінг параметрів та результату виклику.

Такий інтеграційний патерн має досить великий набір недоліків, але дуже широко використовується практично через свою простоту. Швидкість реалізації підкуповує і змушує застосовувати його знову і знову в умовах термінів, що «палають», записуючи рішення в технічний борг. Але буває і так, що недосвідчені розробники застосовують його несвідомо, просто не здогадуючись про негативні наслідки.

Окрім найбільш очевидного підвищення зв'язності підсистем, є й менш явні проблеми з «розтягуванням» та «розтягуванням» транзакцій. Справді, якщо бізнес-логіка вносить якісь зміни, тоді не обійтися без транзакцій, а транзакції, у свою чергу, блокують певні ресурси додатків, які торкаються цих змін. Тобто, поки одна підсистема не дочекається відповіді від іншої, вона не зможе завершити транзакцію і зняти блокування. Це суттєво підвищує ризик виникнення різноманітних ефектів:

втрачається чуйність системи, користувачі довго чекають відповідей на запити;
сервер взагалі перестає відповідати на запити користувачів через переповнений пул потоків: більшість потоків «встали» на блокуванні ресурсу, зайнятого транзакцією;
починають з'являтися дедлоки: ймовірність їх появи залежить від тривалості транзакцій, кількості залученої в транзакцію бізнес-логіки і блокувань;
з'являються помилки закінчення таймууту транзакції;
сервер «падає» по OutOfMemory, якщо завдання вимагає обробки та зміни великих обсягів даних, а наявність синхронних інтеграцій сильно ускладнює подрібнення обробки більш «легкі» транзакції.

З архітектурної точки зору застосування блокуючих викликів під час інтеграції призводить до втрати управління якістю окремих підсистем: неможливо забезпечити цільові показники якості однієї підсистеми у відриві від показників якості іншої підсистеми. Якщо підсистеми розробляються різними командами, то це велика проблема.

Все стає ще цікавіше, якщо підсистеми, що інтегруються, знаходяться в різних додатках і потрібно внести синхронні зміни з двох сторін. Як забезпечити транзакційність цих змін?

Якщо зміни вносяться роздільними транзакціями, тоді потрібно забезпечити надійну обробку винятків та компенсації, а це повністю нівелює основну перевагу синхронних інтеграцій – простоту.

На думку також спадають розподілені транзакції, але ми їх не використовуємо у своїх рішеннях: складно забезпечити надійність.

«Сага» як вирішення проблеми транзакцій

Зі зростанням популярності мікросервісів все більшу затребуваність набуває Шаблон саги.

Даний шаблон відмінно вирішує зазначені вище проблеми довгих транзакцій, а також розширює можливості керування станом системи з боку бізнес-логіки: компенсація після невдалої транзакції може не відкочувати систему у вихідний стан, а забезпечувати альтернативний маршрут обробки даних. Це також дозволяє не повторювати успішно завершені кроки обробки даних під час повторних спроб довести процес до «хорошого» фіналу.

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

Стосовно наших бізнес-процесів у стилі BPM імплементувати «Сагі» виявляється дуже легко: окремі кроки «Сагі» можуть бути задані у вигляді активностей усередині бізнес-процесу, а персистентний стан бізнес-процесу визначає, зокрема, внутрішній стан «Сагі». Тобто нам не потрібно жодного додаткового координаційного механізму. Потрібен лише брокер повідомлень з підтримкою «at least once» гарантій як транспорт.

Але й таке рішення має свою «ціну»:

бізнес-логіка стає складнішою: потрібно відпрацьовувати компенсації;
потрібно відмовитися від повної consistency, що може бути особливо чутливим для монолітних систем;
трохи ускладнюється архітектура, з'являється додаткова потреба у брокері повідомлень;
будуть потрібні додаткові засоби моніторингу та адміністрування (хоча в цілому це навіть добре: якість обслуговування системи підвищиться).

Для монолітних систем виправданість використання «Саг» негаразд очевидна. Для мікросервісів та інших SOA, де, найімовірніше, вже є брокер, а full consistency принесена в жертву ще на старті проекту, користь від використання цього шаблону може значно переважити недоліки, особливо за наявності зручної API на рівні бізнес-логіки.

Інкапсуляція бізнес-логіки у мікросервісах

Коли ми почали експериментувати з мікросервісами, постало резонне питання: куди поміщати доменну бізнес-логіку щодо сервісу, що забезпечує персистенцію доменних даних?

При погляді на архітектуру різних BPMS може здатися розумним відокремити бізнес-логіку від персистенції: створити шар платформних та доменно-незалежних мікросервісів, що формують середовище та контейнер для виконання доменної бізнес-логіки, а персистенцію доменних даних оформити окремим шаром з дуже простих та легковажних мікросервісів. Бізнес-процеси у разі виконують оркестрування сервісів шару персистенції.

Такий підхід має дуже великий плюс: можна скільки завгодно нарощувати функціональність платформи, і «товстіти» від цього буде лише відповідний шар платформних мікросервісів. Бізнес-процеси з будь-якого домену відразу отримують можливість використовувати нову функціональність платформи, як тільки її буде оновлено.

Більш детальне опрацювання виявило суттєві недоліки такого підходу:

платформний сервіс, що виконує бізнес-логіку відразу багатьох доменів, несе у собі великі ризики як єдина точка відмови. Часті зміни бізнес-логіки підвищують ризик виникнення помилок, що призводять до збоїв, що поширюються на всю систему;
проблеми продуктивності: бізнес-логіка працює зі своїми даними через вузький та повільний інтерфейс:
- дані зайвий раз маршалитимуться і прокачуватимуться через мережевий стек;
- доменний сервіс часто віддаватиме більше даних, ніж потрібно бізнес-логіці для обробки, через недостатні можливості параметризації запитів на рівні зовнішнього API сервісу;
- кілька незалежних частин бізнес-логіки можуть повторно перезапитувати ті самі дані для обробки (можна пом'якшити цю проблему додаванням сесійних компонентів, що кешують дані, але це додатково ускладнює архітектуру та створює проблеми актуальності даних та інвалідності кешу);
проблеми транзакційності:
- бізнес-процеси з персистентним станом, зберіганням якого займається платформний сервіс, не погоджуються з доменними даними, і простих шляхів вирішення цієї проблеми не передбачається;
- винесення блокування доменних даних за межі транзакції: якщо доменній бізнес-логіці потрібно внести зміни, попередньо виконавши перевірку коректності актуальних даних, потрібно виключити можливість конкурентної зміни даних, що обробляються. Зовнішнє блокування даних може допомогти вирішити завдання, але таке рішення несе додаткові ризики і знижує загальну надійність системи;
додаткові складності при оновленні: у ряді випадків оновлювати сервіс персистенції та бізнес-логіку потрібно синхронно або у строгій послідовності.

Зрештою довелося повернутися до витоків: інкапсулювати доменні дані та доменну бізнес-логіку в один мікросервіс. Такий підхід спрощує сприйняття мікросервісу як цілісного компонента у складі системи та не породжує перераховані вище проблеми. Це також дається не безкоштовно:

потрібна стандартизація API для взаємодії з бізнес-логікою (зокрема, для забезпечення активності користувачів у складі бізнес-процесів) і API-платформних сервісів; потрібне більш уважне ставлення до зміни API, прямої та зворотної сумісності;
потрібне додавання додаткових runtime-бібліотек для забезпечення функціонування бізнес-логіки у складі кожного такого мікросервісу, і це породжує нові вимоги до таких бібліотек: легковагість та мінімум транзитивних залежностей;
розробникам бізнес-логіки необхідно стежити за версіями бібліотек: якщо якийсь мікросервіс давно не доопрацьовували, то в ньому, швидше за все, застаріла версія бібліотек. Це може стати несподіваною перешкодою для додавання нової фічі та може вимагати міграції старої бізнес-логіки такого сервісу на нові версії бібліотек, якщо між версіями були несумісні зміни.

Шар платформних сервісів у такій архітектурі також присутній, але цей шар формує вже не контейнер для виконання доменної бізнес-логіки, а лише її оточення, надаючи допоміжні «платформні» функції. Такий шар необхідний як збереження легковажності доменних мікросервісів, а й у централізації управління.

Наприклад, активності користувача в бізнес-процесах породжують завдання. Однак, працюючи із завданнями, користувач повинен бачити завдання з усіх доменів у загальному списку, а отже, має бути відповідний платформний сервіс реєстрації завдань, очищений від доменної бізнес-логіки. Зберегти інкапсуляцію бізнес-логіки в такому контексті є досить проблематичним, і це ще один компроміс даної архітектури.

Інтеграція бізнес-процесів очима прикладного розробника

Як уже сказано вище, прикладний розробник має бути абстрагований від технічних та інженерних особливостей реалізації взаємодії кількох додатків, щоб можна було розраховувати на хорошу продуктивність розробки.

Спробуємо вирішити досить непросте інтеграційне завдання, спеціально вигадане для статті. Це буде «ігрова» задача за участю трьох додатків, де кожна з них визначає певне доменне ім'я: «app1», «app2», «app3».

Усередині кожної програми запускаються бізнес-процеси, які починають «грати в м'яч» через інтеграційну шину. У ролі м'яча виступатимуть повідомлення з іменем Ball.

Правила гри:

перший гравець – ініціатор. Він запрошує інших гравців у гру, починає гру і може її будь-якої миті закінчити;
інші гравці заявляють про свою участь у грі, «знайомляться» один з одним та першим гравцем;
прийнявши м'яч, гравець вибирає іншого гравця, що бере участь, і передає йому м'яч. Ведеться підрахунок загальної кількості передач;
кожен гравець має «енергію», яка зменшується з кожною передачею м'яча цим гравцем. Після закінчення енергії гравець вибуває з гри, заявляючи про свій відхід;
якщо гравець залишився один, він відразу заявляє про звільнення;
коли всі гравці вибувають, перший гравець заявляє про завершення гри. Якщо він вибув з гри раніше, залишається стежити за грою, щоб завершити її.

Для вирішення цього завдання я користуюся нашим DSL для бізнес-процесів, що дозволяє описати логіку на Kotlin компактно, з мінімумом бойлерплейту.

У додатку app1 буде працювати бізнес-процес першого гравця (він ініціатор гри):

class InitialPlayer

import ru.krista.bpm.ProcessInstance
import ru.krista.bpm.runtime.ProcessImpl
import ru.krista.bpm.runtime.constraint.UniqueConstraints
import ru.krista.bpm.runtime.dsl.processModel
import ru.krista.bpm.runtime.dsl.taskOperation
import ru.krista.bpm.runtime.instance.MessageSendInstance

data class PlayerInfo(val name: String, val domain: String, val id: String)

class PlayersList : ArrayList<PlayerInfo>()

// Это класс экземпляра процесса: инкапсулирует его внутреннее состояние
class InitialPlayer : ProcessImpl<InitialPlayer>(initialPlayerModel) {
    var playerName: String by persistent("Player1")
    var energy: Int by persistent(30)
    var players: PlayersList by persistent(PlayersList())
    var shotCounter: Int = 0
}

// Это декларация модели процесса: создается один раз, используется всеми
// экземплярами процесса соответствующего класса
val initialPlayerModel = processModel<InitialPlayer>(name = "InitialPlayer",
                                                     version = 1) {

    // По правилам, первый игрок является инициатором игры и должен быть единственным
    uniqueConstraint = UniqueConstraints.singleton

    // Объявляем активности, из которых состоит бизнес-процесс
    val sendNewGameSignal = signal<String>("NewGame")
    val sendStopGameSignal = signal<String>("StopGame")
    val startTask = humanTask("Start") {
        taskOperation {
            processCondition { players.size > 0 }
            confirmation { "Подключилось ${players.size} игроков. Начинаем?" }
        }
    }
    val stopTask = humanTask("Stop") {
        taskOperation {}
    }
    val waitPlayerJoin = signalWait<String>("PlayerJoin") { signal ->
        players.add(PlayerInfo(
                signal.data!!,
                signal.sender.domain,
                signal.sender.processInstanceId))
        println("... join player ${signal.data} ...")
    }
    val waitPlayerOut = signalWait<String>("PlayerOut") { signal ->
        players.remove(PlayerInfo(
                signal.data!!,
                signal.sender.domain,
                signal.sender.processInstanceId))
        println("... player ${signal.data} is out ...")
    }
    val sendPlayerOut = signal<String>("PlayerOut") {
        signalData = { playerName }
    }
    val sendHandshake = messageSend<String>("Handshake") {
        messageData = { playerName }
        activation = {
            receiverDomain = process.players.last().domain
            receiverProcessInstanceId = process.players.last().id
        }
    }
    val throwStartBall = messageSend<Int>("Ball") {
        messageData = { 1 }
        activation = { selectNextPlayer() }
    }
    val throwBall = messageSend<Int>("Ball") {
        messageData = { shotCounter + 1 }
        activation = { selectNextPlayer() }
        onEntry { energy -= 1 }
    }
    val waitBall = messageWaitData<Int>("Ball") {
        shotCounter = it
    }

    // Теперь конструируем граф процесса из объявленных активностей
    startFrom(sendNewGameSignal)
            .fork("mainFork") {
                next(startTask)
                next(waitPlayerJoin).next(sendHandshake).next(waitPlayerJoin)
                next(waitPlayerOut)
                        .branch("checkPlayers") {
                            ifTrue { players.isEmpty() }
                                    .next(sendStopGameSignal)
                                    .terminate()
                            ifElse().next(waitPlayerOut)
                        }
            }
    startTask.fork("afterStart") {
        next(throwStartBall)
                .branch("mainLoop") {
                    ifTrue { energy < 5 }.next(sendPlayerOut).next(waitBall)
                    ifElse().next(waitBall).next(throwBall).loop()
                }
        next(stopTask).next(sendStopGameSignal)
    }

    // Навешаем на активности дополнительные обработчики для логирования
    sendNewGameSignal.onExit { println("Let's play!") }
    sendStopGameSignal.onExit { println("Stop!") }
    sendPlayerOut.onExit { println("$playerName: I'm out!") }
}

private fun MessageSendInstance<InitialPlayer, Int>.selectNextPlayer() {
    val player = process.players.random()
    receiverDomain = player.domain
    receiverProcessInstanceId = player.id
    println("Step ${process.shotCounter + 1}: " +
            "${process.playerName} >>> ${player.name}")
}

Окрім виконання бізнес-логіки, наведений код вміє видавати об'єктну модель бізнес-процесу, яка може бути візуалізована у вигляді діаграми. Візуалізатор ми поки не реалізували, тому довелося витратити трохи часу на малювання (тут я злегка спростив BPMN нотацію щодо використання гейтів, щоб покращити узгодженість діаграми з наведеним кодом):

Додаток app2 буде включати бізнес-процес іншого гравця:

class RandomPlayer

import ru.krista.bpm.ProcessInstance
import ru.krista.bpm.runtime.ProcessImpl
import ru.krista.bpm.runtime.dsl.processModel
import ru.krista.bpm.runtime.instance.MessageSendInstance

data class PlayerInfo(val name: String, val domain: String, val id: String)

class PlayersList: ArrayList<PlayerInfo>()

class RandomPlayer : ProcessImpl<RandomPlayer>(randomPlayerModel) {

    var playerName: String by input(persistent = true, 
                                    defaultValue = "RandomPlayer")
    var energy: Int by input(persistent = true, defaultValue = 30)
    var players: PlayersList by persistent(PlayersList())
    var allPlayersOut: Boolean by persistent(false)
    var shotCounter: Int = 0

    val selfPlayer: PlayerInfo
        get() = PlayerInfo(playerName, env.eventDispatcher.domainName, id)
}

val randomPlayerModel = processModel<RandomPlayer>(name = "RandomPlayer", 
                                                   version = 1) {

    val waitNewGameSignal = signalWait<String>("NewGame")
    val waitStopGameSignal = signalWait<String>("StopGame")
    val sendPlayerJoin = signal<String>("PlayerJoin") {
        signalData = { playerName }
    }
    val sendPlayerOut = signal<String>("PlayerOut") {
        signalData = { playerName }
    }
    val waitPlayerJoin = signalWaitCustom<String>("PlayerJoin") {
        eventCondition = { signal ->
            signal.sender.processInstanceId != process.id 
                && !process.players.any { signal.sender.processInstanceId == it.id}
        }
        handler = { signal ->
            players.add(PlayerInfo(
                    signal.data!!,
                    signal.sender.domain,
                    signal.sender.processInstanceId))
        }
    }
    val waitPlayerOut = signalWait<String>("PlayerOut") { signal ->
        players.remove(PlayerInfo(
                signal.data!!,
                signal.sender.domain,
                signal.sender.processInstanceId))
        allPlayersOut = players.isEmpty()
    }
    val sendHandshake = messageSend<String>("Handshake") {
        messageData = { playerName }
        activation = {
            receiverDomain = process.players.last().domain
            receiverProcessInstanceId = process.players.last().id
        }
    }
    val receiveHandshake = messageWait<String>("Handshake") { message ->
        if (!players.any { message.sender.processInstanceId == it.id}) {
            players.add(PlayerInfo(
                    message.data!!, 
                    message.sender.domain, 
                    message.sender.processInstanceId))
        }
    }
    val throwBall = messageSend<Int>("Ball") {
        messageData = { shotCounter + 1 }
        activation = { selectNextPlayer() }
        onEntry { energy -= 1 }
    }
    val waitBall = messageWaitData<Int>("Ball") {
        shotCounter = it
    }

    startFrom(waitNewGameSignal)
            .fork("mainFork") {
                next(sendPlayerJoin)
                        .branch("mainLoop") {
                            ifTrue { energy < 5 || allPlayersOut }
                                    .next(sendPlayerOut)
                                    .next(waitBall)
                            ifElse()
                                    .next(waitBall)
                                    .next(throwBall)
                                    .loop()
                        }
                next(waitPlayerJoin).next(sendHandshake).next(waitPlayerJoin)
                next(waitPlayerOut).next(waitPlayerOut)
                next(receiveHandshake).next(receiveHandshake)
                next(waitStopGameSignal).terminate()
            }

    sendPlayerJoin.onExit { println("$playerName: I'm here!") }
    sendPlayerOut.onExit { println("$playerName: I'm out!") }
}

private fun MessageSendInstance<RandomPlayer, Int>.selectNextPlayer() {
    val player = if (process.players.isNotEmpty()) 
        process.players.random() 
    else 
        process.selfPlayer
    receiverDomain = player.domain
    receiverProcessInstanceId = player.id
    println("Step ${process.shotCounter + 1}: " +
            "${process.playerName} >>> ${player.name}")
}

Діаграма:

У додатку app3 зробимо гравця трохи з іншою поведінкою: замість випадкового вибору наступного гравця він діятиме за алгоритмом round-robin:

class RoundRobinPlayer

import ru.krista.bpm.ProcessInstance
import ru.krista.bpm.runtime.ProcessImpl
import ru.krista.bpm.runtime.dsl.processModel
import ru.krista.bpm.runtime.instance.MessageSendInstance

data class PlayerInfo(val name: String, val domain: String, val id: String)

class PlayersList: ArrayList<PlayerInfo>()

class RoundRobinPlayer : ProcessImpl<RoundRobinPlayer>(roundRobinPlayerModel) {

    var playerName: String by input(persistent = true, 
                                    defaultValue = "RoundRobinPlayer")
    var energy: Int by input(persistent = true, defaultValue = 30)
    var players: PlayersList by persistent(PlayersList())
    var nextPlayerIndex: Int by persistent(-1)
    var allPlayersOut: Boolean by persistent(false)
    var shotCounter: Int = 0

    val selfPlayer: PlayerInfo
        get() = PlayerInfo(playerName, env.eventDispatcher.domainName, id)
}

val roundRobinPlayerModel = processModel<RoundRobinPlayer>(
        name = "RoundRobinPlayer", 
        version = 1) {

    val waitNewGameSignal = signalWait<String>("NewGame")
    val waitStopGameSignal = signalWait<String>("StopGame")
    val sendPlayerJoin = signal<String>("PlayerJoin") {
        signalData = { playerName }
    }
    val sendPlayerOut = signal<String>("PlayerOut") {
        signalData = { playerName }
    }
    val waitPlayerJoin = signalWaitCustom<String>("PlayerJoin") {
        eventCondition = { signal ->
            signal.sender.processInstanceId != process.id 
                && !process.players.any { signal.sender.processInstanceId == it.id}
        }
        handler = { signal ->
            players.add(PlayerInfo(
                    signal.data!!, 
                    signal.sender.domain, 
                    signal.sender.processInstanceId))
        }
    }
    val waitPlayerOut = signalWait<String>("PlayerOut") { signal ->
        players.remove(PlayerInfo(
                signal.data!!, 
                signal.sender.domain, 
                signal.sender.processInstanceId))
        allPlayersOut = players.isEmpty()
    }
    val sendHandshake = messageSend<String>("Handshake") {
        messageData = { playerName }
        activation = {
            receiverDomain = process.players.last().domain
            receiverProcessInstanceId = process.players.last().id
        }
    }
    val receiveHandshake = messageWait<String>("Handshake") { message ->
        if (!players.any { message.sender.processInstanceId == it.id}) {
            players.add(PlayerInfo(
                    message.data!!, 
                    message.sender.domain, 
                    message.sender.processInstanceId))
        }
    }
    val throwBall = messageSend<Int>("Ball") {
        messageData = { shotCounter + 1 }
        activation = { selectNextPlayer() }
        onEntry { energy -= 1 }
    }
    val waitBall = messageWaitData<Int>("Ball") {
        shotCounter = it
    }

    startFrom(waitNewGameSignal)
            .fork("mainFork") {
                next(sendPlayerJoin)
                        .branch("mainLoop") {
                            ifTrue { energy < 5 || allPlayersOut }
                                    .next(sendPlayerOut)
                                    .next(waitBall)
                            ifElse()
                                    .next(waitBall)
                                    .next(throwBall)
                                    .loop()
                        }
                next(waitPlayerJoin).next(sendHandshake).next(waitPlayerJoin)
                next(waitPlayerOut).next(waitPlayerOut)
                next(receiveHandshake).next(receiveHandshake)
                next(waitStopGameSignal).terminate()
            }

    sendPlayerJoin.onExit { println("$playerName: I'm here!") }
    sendPlayerOut.onExit { println("$playerName: I'm out!") }
}

private fun MessageSendInstance<RoundRobinPlayer, Int>.selectNextPlayer() {
    var idx = process.nextPlayerIndex + 1
    if (idx >= process.players.size) {
        idx = 0
    }
    process.nextPlayerIndex = idx
    val player = if (process.players.isNotEmpty()) 
        process.players[idx] 
    else 
        process.selfPlayer
    receiverDomain = player.domain
    receiverProcessInstanceId = player.id
    println("Step ${process.shotCounter + 1}: " +
            "${process.playerName} >>> ${player.name}")
}

В іншому поведінка гравця не відрізняється від попередньої, тож діаграма не змінюється.

Тепер потрібний тест, щоб усе це запускати. Наведу лише код самого тесту, щоб не захаращувати статтю бойлерплейтом (насправді я скористався тестовим оточенням, створеним раніше для тестування інтеграції інших бізнес-процесів):

testGame()

@Test
public void testGame() throws InterruptedException {
    String pl2 = startProcess(app2, "RandomPlayer", playerParams("Player2", 20));
    String pl3 = startProcess(app2, "RandomPlayer", playerParams("Player3", 40));
    String pl4 = startProcess(app3, "RoundRobinPlayer", playerParams("Player4", 25));
    String pl5 = startProcess(app3, "RoundRobinPlayer", playerParams("Player5", 35));
    String pl1 = startProcess(app1, "InitialPlayer");
    // Теперь нужно немного подождать, пока игроки "познакомятся" друг с другом.
    // Ждать через sleep - плохое решение, зато самое простое. 
    // Не делайте так в серьезных тестах!
    Thread.sleep(1000);
    // Запускаем игру, закрывая пользовательскую активность
    assertTrue(closeTask(app1, pl1, "Start"));
    app1.getWaiting().waitProcessFinished(pl1);
    app2.getWaiting().waitProcessFinished(pl2);
    app2.getWaiting().waitProcessFinished(pl3);
    app3.getWaiting().waitProcessFinished(pl4);
    app3.getWaiting().waitProcessFinished(pl5);
}

private Map<String, Object> playerParams(String name, int energy) {
    Map<String, Object> params = new HashMap<>();
    params.put("playerName", name);
    params.put("energy", energy);
    return params;
}

Запускаємо тест, дивимося лог:

вихід консолі

Взята блокировка ключа lock://app1/process/InitialPlayer
Let's play!
Снята блокировка ключа lock://app1/process/InitialPlayer
Player2: I'm here!
Player3: I'm here!
Player4: I'm here!
Player5: I'm here!
... join player Player2 ...
... join player Player4 ...
... join player Player3 ...
... join player Player5 ...
Step 1: Player1 >>> Player3
Step 2: Player3 >>> Player5
Step 3: Player5 >>> Player3
Step 4: Player3 >>> Player4
Step 5: Player4 >>> Player3
Step 6: Player3 >>> Player4
Step 7: Player4 >>> Player5
Step 8: Player5 >>> Player2
Step 9: Player2 >>> Player5
Step 10: Player5 >>> Player4
Step 11: Player4 >>> Player2
Step 12: Player2 >>> Player4
Step 13: Player4 >>> Player1
Step 14: Player1 >>> Player4
Step 15: Player4 >>> Player3
Step 16: Player3 >>> Player1
Step 17: Player1 >>> Player2
Step 18: Player2 >>> Player3
Step 19: Player3 >>> Player1
Step 20: Player1 >>> Player5
Step 21: Player5 >>> Player1
Step 22: Player1 >>> Player2
Step 23: Player2 >>> Player4
Step 24: Player4 >>> Player5
Step 25: Player5 >>> Player3
Step 26: Player3 >>> Player4
Step 27: Player4 >>> Player2
Step 28: Player2 >>> Player5
Step 29: Player5 >>> Player2
Step 30: Player2 >>> Player1
Step 31: Player1 >>> Player3
Step 32: Player3 >>> Player4
Step 33: Player4 >>> Player1
Step 34: Player1 >>> Player3
Step 35: Player3 >>> Player4
Step 36: Player4 >>> Player3
Step 37: Player3 >>> Player2
Step 38: Player2 >>> Player5
Step 39: Player5 >>> Player4
Step 40: Player4 >>> Player5
Step 41: Player5 >>> Player1
Step 42: Player1 >>> Player5
Step 43: Player5 >>> Player3
Step 44: Player3 >>> Player5
Step 45: Player5 >>> Player2
Step 46: Player2 >>> Player3
Step 47: Player3 >>> Player2
Step 48: Player2 >>> Player5
Step 49: Player5 >>> Player4
Step 50: Player4 >>> Player2
Step 51: Player2 >>> Player5
Step 52: Player5 >>> Player1
Step 53: Player1 >>> Player5
Step 54: Player5 >>> Player3
Step 55: Player3 >>> Player5
Step 56: Player5 >>> Player2
Step 57: Player2 >>> Player1
Step 58: Player1 >>> Player4
Step 59: Player4 >>> Player1
Step 60: Player1 >>> Player4
Step 61: Player4 >>> Player3
Step 62: Player3 >>> Player2
Step 63: Player2 >>> Player5
Step 64: Player5 >>> Player4
Step 65: Player4 >>> Player5
Step 66: Player5 >>> Player1
Step 67: Player1 >>> Player5
Step 68: Player5 >>> Player3
Step 69: Player3 >>> Player4
Step 70: Player4 >>> Player2
Step 71: Player2 >>> Player5
Step 72: Player5 >>> Player2
Step 73: Player2 >>> Player1
Step 74: Player1 >>> Player4
Step 75: Player4 >>> Player1
Step 76: Player1 >>> Player2
Step 77: Player2 >>> Player5
Step 78: Player5 >>> Player4
Step 79: Player4 >>> Player3
Step 80: Player3 >>> Player1
Step 81: Player1 >>> Player5
Step 82: Player5 >>> Player1
Step 83: Player1 >>> Player4
Step 84: Player4 >>> Player5
Step 85: Player5 >>> Player3
Step 86: Player3 >>> Player5
Step 87: Player5 >>> Player2
Step 88: Player2 >>> Player3
Player2: I'm out!
Step 89: Player3 >>> Player4
... player Player2 is out ...
Step 90: Player4 >>> Player1
Step 91: Player1 >>> Player3
Step 92: Player3 >>> Player1
Step 93: Player1 >>> Player4
Step 94: Player4 >>> Player3
Step 95: Player3 >>> Player5
Step 96: Player5 >>> Player1
Step 97: Player1 >>> Player5
Step 98: Player5 >>> Player3
Step 99: Player3 >>> Player5
Step 100: Player5 >>> Player4
Step 101: Player4 >>> Player5
Player4: I'm out!
... player Player4 is out ...
Step 102: Player5 >>> Player1
Step 103: Player1 >>> Player3
Step 104: Player3 >>> Player1
Step 105: Player1 >>> Player3
Step 106: Player3 >>> Player5
Step 107: Player5 >>> Player3
Step 108: Player3 >>> Player1
Step 109: Player1 >>> Player3
Step 110: Player3 >>> Player5
Step 111: Player5 >>> Player1
Step 112: Player1 >>> Player3
Step 113: Player3 >>> Player5
Step 114: Player5 >>> Player3
Step 115: Player3 >>> Player1
Step 116: Player1 >>> Player3
Step 117: Player3 >>> Player5
Step 118: Player5 >>> Player1
Step 119: Player1 >>> Player3
Step 120: Player3 >>> Player5
Step 121: Player5 >>> Player3
Player5: I'm out!
... player Player5 is out ...
Step 122: Player3 >>> Player5
Step 123: Player5 >>> Player1
Player5: I'm out!
Step 124: Player1 >>> Player3
... player Player5 is out ...
Step 125: Player3 >>> Player1
Step 126: Player1 >>> Player3
Player1: I'm out!
... player Player1 is out ...
Step 127: Player3 >>> Player3
Player3: I'm out!
Step 128: Player3 >>> Player3
... player Player3 is out ...
Player3: I'm out!
Stop!
Step 129: Player3 >>> Player3
Player3: I'm out!

З усього цього можна зробити кілька важливих висновків:

за наявності необхідних інструментів прикладні розробники можуть створювати інтеграційні взаємодії між програмами без відриву від бізнес-логіки;
складність (complexity) інтеграційної задачі, що вимагає інженерних компетенцій, можна приховати всередині фреймворку, якщо спочатку закласти це в архітектуру фреймворку. Труднощі задачі (difficulty) приховати не вийде, тому розв'язання важкої задачі в коді виглядатиме відповідно;
При розробці інтеграційної логіки обов'язково потрібно враховувати еventually consistency та відсутність лінеаризуемості зміни стану всіх учасників інтеграції. Це змушує ускладнювати логіку, щоб зробити її нечутливою до виникнення зовнішніх подій. У нашому прикладі гравець змушений брати участь у грі вже після того, як він заявить про вихід із гри: інші гравці продовжуватимуть передавати йому м'яч, поки інформація про його вихід не дійде та не обробиться всіма учасниками. Ця логіка не випливає із правил гри і є компромісним рішенням у рамках обраної архітектури.

Далі поговоримо про різні тонкощі нашого рішення, компроміси та інші моменти.

Всі повідомлення – в одній черзі

Всі інтегровані програми працюють з однією інтеграційною шиною, яка представлена у вигляді зовнішнього брокера, однієї черги BPMQueue – для повідомлень та одного топіка BPMTopic – для сигналів (подій). Пропускати всі повідомлення через одну чергу є компромісом. На рівні бізнес-логіки тепер можна вводити скільки завгодно нових типів повідомлень, не вносячи змін до структури системи. Це значне спрощення, але воно несе в собі певні ризики, які в контексті наших типових завдань здалися нам не такими вже й значними.

Однак тут є одна тонкість: кожен додаток відфільтровує свої повідомлення з черги ще на вході, на ім'я свого домену. Також домен може бути зазначений і в сигналах, якщо потрібно обмежити область видимості сигналу одним єдиним додатком. Це має збільшити пропускну спроможність шини, але бізнес-логіка тепер має оперувати іменами доменів: для адресації повідомлень обов'язково, для сигналів бажано.

Забезпечення надійності інтеграційної шини

Надійність складається з кількох моментів:

вибраний брокер повідомлень – критично важливий компонент архітектури та єдина точка відмови: він має бути досить стійким до відмови. Слід використовувати лише перевірені часом реалізації, з гарною підтримкою та великим ком'юніті;
необхідно забезпечити високу доступність брокера повідомлень, для чого він повинен бути фізично відокремлений від інтегрованих додатків (високу доступність додатків із прикладною бізнес-логікою забезпечити значно складніше та дорожче);
Брокер зобов'язаний забезпечити "at least once" гарантії доставки. Це є обов'язковою вимогою для надійної роботи інтеграційної шини. У гарантіях рівня «exactly once» немає необхідності: бізнес-процеси, як правило, не чутливі до повторного надходження повідомлень чи подій, а в особливих завданнях, де це важливо, простіше додати додаткову перевірку в бізнес-логіку, ніж постійно використовувати досить дорогі гарантії;
надсилання повідомлень та сигналів необхідно залучати до загальної транзакції зі зміною стану бізнес-процесів та доменних даних. Переважним варіантом буде використання патерну Transactional Outboxале воно вимагатиме наявності додаткової таблиці в базі і ретранслятора. У JEE-програмах можна спростити цей момент з використанням локального JTA-менеджера, але підключення до вибраного брокера має вміти працювати в режимі XA;
обробники вхідних повідомлень та подій також повинні працювати з транзакцією зміни стану бізнес-процесу: якщо така транзакція відкочується, то і прийом повідомлення має бути скасовано;
повідомлення, які не вдалося доставити через помилки, потрібно складати в окреме сховище DLQ (Dead Letter Queue). Ми для цього створили окремий платформний мікросервіс, який зберігає такі повідомлення у своєму сховищі, індексує їх за атрибутами (для швидкого угруповання та пошуку), і виставляє API для перегляду, повторного надсилання на адресу призначення, видалення повідомлень. Адміністратори системи можуть працювати із цим сервісом через свій веб-інтерфейс;
у налаштуваннях брокера потрібно підлаштувати кількість повторних спроб доставки та затримки між доставками, щоб зменшити ймовірність попадання повідомлень у DLQ (обчислити оптимальні параметри практично нереально, але можна діяти емпірично та підлаштовувати їх під час експлуатації);
сховище DLQ повинно безперервно моніторитися, і система моніторингу повинна сповіщати адміністраторів системи, щоб при появі недоставлених повідомлень реагувати якнайшвидше. Це дозволить зменшити «зону поразки» збою, що виник, або помилки бізнес-логіки;
інтеграційна шина повинна бути нечутлива до тимчасової відсутності додатків: підписки на топік повинні бути тривалими, а доменне ім'я додатка має бути унікальним, щоб за час відсутності додатка його повідомлення з черги не спробував опрацювати хтось інший.

Забезпечення потокобезпеки бізнес-логіки

Одному екземпляру бізнес-процесу може надійти відразу кілька повідомлень і подій, обробка яких запуститься паралельно. У той же час для прикладного розробника все має бути простим і потокобезпечним.

Бізнес-логіка процесу обробляє кожну зовнішню подію, що впливає цей бізнес-процес, окремо. Такими подіями можуть бути:

запуск екземпляра бізнес-процесу;
дія користувача, що стосується активності всередині бізнес-процесу;
надходження повідомлення або сигналу, на який підписано екземпляр бізнес-процесу;
спрацьовування таймера, встановленого екземпляром бізнес-процесу;
керуюча дія через API (наприклад, аварійне переривання процесу).

Кожна така подія може змінити стан екземпляра бізнес-процесу: можуть завершитися одні активності та розпочатися інші, можуть змінитися значення персистентних властивостей. Закриття будь-якої активності може призвести до активації однієї або кількох таких активностей. Ті, у свою чергу, можуть зупинитися на очікуванні інших подій або якщо їм не потрібні жодні додаткові дані, можуть завершитися в тій же транзакції. Перед закриттям транзакції новий стан бізнес-процесу зберігається в БД, де чекатиме настання наступної зовнішньої події.

Персистентні дані бізнес-процесу, збережені в реляційну БД є дуже зручною точкою синхронізації обробки, якщо використовувати SELECT FOR UPDATE. Якщо одній транзакції вдалося отримати стан бізнес-процесу з бази для його зміни, то ніяка інша транзакція паралельно не зможе отримати цей стан для іншої зміни, а після завершення першої транзакції друга гарантовано отримає вже змінений стан.

Використовуючи песимістичні блокування на стороні СУБД, ми виконуємо всі необхідні вимоги ACID, а також зберігаємо можливість масштабування програми з бізнес-логікою шляхом збільшення кількості запущених екземплярів.

Однак песимістичні блокування загрожують нам дідлоками, а значить, SELECT FOR UPDATE все-таки варто обмежити деяким розумним таймом на випадок виникнення дедлок на якихось кричачих кейсах в бізнес-логіці.

Ще одна проблема – синхронізація старту бізнес-процесу. Поки немає екземпляра бізнес-процесу, немає його стану в базі, тому описаний метод не підійде. Якщо потрібно забезпечити унікальність екземпляра бізнес-процесу в певному скоупі, тоді буде потрібно деякий об'єкт синхронізації, асоційований з класом процесу та відповідним скоупом. Для вирішення цієї проблеми ми використовуємо інший механізм блокування, що дозволяє взяти блокування довільного ресурсу, заданого ключем у форматі URI, через зовнішній сервіс.

У наших прикладах бізнес-процес InitialPlayer містить оголошення

uniqueConstraint = UniqueConstraints.singleton

Тому в лозі присутні повідомлення про взяття та звільнення блокування відповідного ключа. За іншими бізнес-процесами таких повідомлень немає: uniqueConstraint не заданий.

Проблеми бізнес-процесів із персистентним станом

Іноді наявність персистентного стану як допомагає, а й дуже заважає у створенні.
Проблеми починаються, коли потрібно внести зміни до бізнес-логіки та/або моделі бізнес-процесу. Не будь-яка така зміна виявляється сумісною зі старим станом бізнес-процесів. Якщо в базі даних є багато «живих» екземплярів, тоді внесення несумісних змін може завдати безліч неприємностей, з якими ми часто стикалися при використанні jBPM.

Залежно від глибини змін можна діяти двома шляхами:

створити новий тип бізнес-процесу, щоб не вносити несумісні зміни до старого, і використовувати його замість старого при запуску нових екземплярів. Старі екземпляри продовжуватимуть працювати «по-старому»;
мігрувати персистентний стан бізнес-процесів під час оновлення бізнес-логіки.

Перший шлях більш простий, але має свої обмеження та недоліки, наприклад:

дублювання бізнес-логіки у багатьох моделях бізнес-процесів; збільшення обсягу бізнес-логіки;
часто потрібен моментальний перехід на нову бізнес-логіку (у частині інтеграційних завдань – майже завжди);
розробник не знає, коли можна видаляти застарілі моделі.

На практиці ми використовуємо обидва підходи, але ухвалили низку рішень, щоб спростити собі життя:

в базі даних персистентний стан бізнес-процесу зберігається в легко читаному та легко оброблюваному вигляді: у рядку формату JSON. Це дозволяє виконувати міграції як усередині програми, так і зовні. В крайньому випадку можна і ручками підправити (особливо корисно у розробці під час налагодження);
інтеграційна бізнес-логіка не використовує імена бізнес-процесів, щоб у будь-який момент можна було замінити реалізацію одного з процесів, що беруть участь, на нову, з новим ім'ям (наприклад, «InitialPlayerV2»). Зв'язування відбувається через імена повідомлень та сигналів;
модель процесу має номер версії, який ми збільшуємо, якщо вносимо до цієї моделі несумісні зміни, і цей номер зберігається разом із станом екземпляра процесу;
персистентний стан процесу вираховується з бази спочатку зручну об'єктну модель, з якою може працювати процедура міграції, якщо змінився номер версії моделі;
процедура міграції розміщується поруч із бізнес-логікою та викликається «ліниво» для кожного екземпляра бізнес-процесу в момент його відновлення з бази;
якщо потрібно мігрувати стан усіх екземплярів процесу оперативно та синхронно, застосовуються більш класичні рішення щодо міграції БД, але там доводиться працювати з JSON.

Чи потрібний ще один фреймворк для бізнес-процесів?

Описані у статті рішення дозволили нам помітно спростити собі життя, розширити коло питань, які вирішуються лише на рівні прикладної розробки, зробити привабливішими ідеї виділення бізнес-логіки в мікросервіси. Для цього було зроблено багато роботи, створено дуже «легковажний» фреймворк для бізнес-процесів, а також службові компоненти для вирішення зазначених проблем у контексті широкого кола прикладних завдань. У нас є бажання поділитися цими результатами, розробити спільні компоненти у відкритий доступ під вільною ліцензією. Це вимагатиме певних зусиль та часу. Розуміння затребуваності таких рішень може стати для нас додатковим стимулом. У запропонованій статті мало уваги приділено можливостям самого фреймворку, але деякі з них видно з наведених прикладів. Якщо ми таки опублікуємо свій фреймворк, йому буде присвячена окрема стаття. А поки будемо вдячні, якщо залишите невеликий фідбек, відповівши на запитання:

Тільки зареєстровані користувачі можуть брати участь в опитуванні. Увійдіть, будь ласка.

чи потрібен ще один фреймворк для бізнес-процесів?

18,8%так, давно шукаємо щось подібне3
12,5%цікаво дізнатися про вашу реалізацію побільше, може стати в нагоді2
6,2%використовуємо один із існуючих фреймворків, але подумуємо про заміну1
18,8%використовуємо один з існуючих фреймворків, все влаштовує3
18,8%справляємося без фреймворку3
25,0%пишемо свій4

Проголосували 16 користувачів. Утрималися 7 користувачів.

Джерело: habr.com

Інтеграція у стилі BPM