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

Здравейте, Хабр!

Нашата компания се специализира в разработката на ERP софтуерни решения, в които голям дял заемат трансакционните системи с обширна бизнес логика и документооборот, подобен на СЭД. Съвременните версии на нашите продукти са изградени върху технологии JavaEE, но активно експериментираме и с микросервизи. Една от най-проблемните области при тези решения е интеграцията на различни подсистеми, свързани с близки домейни. Задачите по интеграция винаги са ни причинявали големи главоболия, независимо от архитектурните стилове, технологичните стека и фреймуърките, които използваме; обаче, напоследък има напредък в решаването на тези предизвикателства.

В предложената статия ще споделя опита и архитектурните изисквания на НПО "Криста" в посочената област. Ще разгледаме и пример за просто решение на интеграционна задача от гледна точка на приложния разработчик и ще разберем какво се крие зад тази простота.

Дисклеймер

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

Каква е връзката с BPM?

За да отговорим на този въпрос, трябва да навлезем малко по-дълбоко в спецификата на приложните задачи на нашите решения. Основната част от бизнес логиката в нашата типична трансакционна система включва въвеждане на данни в БД чрез потребителски интерфейси, ръчна и автоматизирана проверка на тези данни, провеждане на workflow, публикуване в друга система / аналитична база / архив, и генериране на отчети. По този начин, ключовата функция на системата за клиентите е автоматизацията на техните вътрешни бизнес процеси.

За удобство, в комуникацията използваме термина "документ" като някаква абстракция на набор от данни, обединени около общ ключ, към който може да бъде "привързан" определен workflow.
Но какво да правим с интеграционната логика? Нали задачата по интеграция произтича от архитектурата на системата, която е "разделена" на части не по искане на клиента, а под влиянието на съвсем различни фактори:

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

Съществува голямо изкушение да се отдели интеграционната логика от бизнес логиката на основния работен поток, за да не се замърсява бизнес логиката с интеграционни артефакти и да се освободи приложният разработчик от необходимостта да се задълбочава в особеностите на архитектурния ландшафт на системата. Такъв подход има редица предимства, но практиката показва, че той е неефективен:

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

Друг подход е разглеждане на интеграционните взаимодействия като неразделна част от основната бизнес логика и работния процес. За да не се покачат изискванията към квалификацията на приложимите разработчици до небето, създаването на нови интеграционни взаимодействия трябва да се извършва лесно и без усилия, с минимални възможности за избор на начин за решение. Това е по-трудно, отколкото изглежда: инструментът трябва да бъде достатъчно мощен, за да предостави на потребителя необходимото множество възможности за приложение и в същото време да предотврати „самоубийствени“ решения. Съществуват множество въпроси, на които инженерът трябва да отговори в контекста на интеграционните задачи, но за които не трябва да се тревожи приложимият разработчик в своята ежедневна работа: граници на транзакции, консистентност, атомарност, сигурност, мащабируемост, разпределение на натоварвания и ресурси, маршрутизиране, маршалинг, разпространение и превключване на контексти и т.н. Нужно е да се предложат на приложимите разработчици достатъчно прости шаблони решения, в които вече да са скрити отговорите на всички подобни въпроси. Тези шаблони трябва да бъдат достатъчно безопасни: бизнес логиката се променя много често, което увеличава рисковете от допускане на грешки, а цената на грешките трябва да остане на достатъчно ниско ниво.

Но все пак какво общо има BPM с това? Има много варианти за реализиране на работния процес…
Наистина, в нашите решения много популярна е друга реализация на бизнес процесите - чрез декларативно задаване на диаграмми на преминаване на състояния и свързване на обработчици с бизнес логика за преходите. В същото време, състоянието, определящо текущото положение на „документа“ в бизнес процеса, е атрибут на самия „документ“.

Така изглежда процесът в началото на проекта.

Популярността на такова реализиране се дължи на относителната му простота и бързина при създаването на линейни бизнес процеси. Въпреки това, с постоянното усложняване на софтуерните системи автоматизираната част от бизнес процеса се разраства и усложнява. Появява се необходимостта от декомпозиция, повторна употреба на части от процесите, както и от разклоняване на процесите, така че всяко клонче да се изпълнява паралелно. При такива условия инструментът става неудобен, а диаграмата на състоянията губи информативност (интеграционните взаимодействия изобщо не се отразяват на диаграмата).

Така изглежда процесът след няколко итерации на уточняване на изискванията

Изходът от тази ситуация стана интеграцията на движка jBPM в някои продукти с най-сложните бизнес процеси. В краткосрочен план това решение имаше определен успех: появи се възможност за реализиране на сложни бизнес процеси, запазвайки достатъчно информативна и актуална диаграма в нотация BPMN2.

Малка част от сложен бизнес процес

В дългосрочен план решението не оправда очакванията: високата трудоемкост при създаването на бизнес процеси чрез визуални инструменти не позволи постигането на удовлетворителни показатели за продуктивност, а самият инструмент стана един от най-нехаресваните сред разработчиците. Имаше също и оплаквания относно вътрешната структура на движка, които доведоха до появата на множество 'пачове' и 'костури'.

Главното положително нещо от прилагането на jBPM стана осъзнаването на ползите и вредите от наличието на собствено персистентно състояние на инстанцията на бизнес процеса. Също така видяхме възможност за прилагане на процесния подход за реализиране на сложни интеграционни протоколи между различни приложения с използване на асинхронни взаимодействия чрез сигнали и съобщения. Наличието на персистентно състояние играе изключително важна роля в това.

На базата на казаното можем да направим извода: процесният подход в стил BPM ни позволява да решаваме широк спектър от задачи по автоматизация на постояннопроцесно усложняващите се бизнес процеси, хармонично вмествайки интеграционни активности в тези процеси и запазвайки възможността за визуално представяне на реализирания процес в подходяща за това нотация.

Недостатъци на синхронните повиквания като интеграционен модел

Под синхронна интеграция се разбира прост блокиращ повик. Една подсистема играе ролята на сървър и предлага API с необходимия метод. Друга подсистема е клиентска и в необходимия момент извършва повика с очакване на резултата. В зависимост от архитектурата на системата, клиентската и сървърната страна могат да бъдат разположени или в едно приложение и процес, или в различни. Във втория случай е необходимо да се приложи някаква реализация на RPC и да се осигури маршалиране на параметрите и резултата от повика.

Този интеграционен модел има сравнително голям набор от недостатъци, но той се използва много широко на практика поради своята простота. Скоростта на реализация привлича и води до повторно прилагане в условия на "изгарящи" срокове, записвайки решението в техническия дълг. Но понякога неопитни разработчици го прилагат неосъзнато, просто не подозирайки за негативните последици.

Освен най-очевидното повишаване на свързаността на подсистемите, съществуват и по-малко явни проблеми с "разпокъсването" и "разтягането" на транзакциите. Наистина, ако бизнес логиката внася някакви изменения, тогава не може да се избегне без транзакции, а транзакциите, от своя страна, блокират определени ресурси на приложението, засегнати от тези изменения. Тоест, докато една подсистема не получи отговора от друга, тя не може да завърши транзакцията и да освободи блокировките. Това значително увеличава риска от възникване на разнообразни ефекти:

загуба на отзивчивост на системата, потребителите дълго чакат отговори на запитванията;
сървърът изцяло спира да отговаря на потребителските запитвания поради препълнен пул от нишки: повечето нишки "стоират" на блокировка на ресурс, зает от транзакцията;
започват да се появяват мъртви блокажи: вероятността за тяхното възникване силно зависи от продължителността на транзакциите, количеството заета в транзакцията бизнес логика и блокировките;
появяват се грешки за изтичане на времето на транзакцията;
сървърът "пада" по OutOfMemory, ако задачата изисква обработка и променяне на големи обеми данни, а наличието на синхронни интеграции значително затруднява разделянето на обработката на по-"леки" транзакции.

От архитектурна гледна точка, използването на блокиращи извиквания при интеграцията води до загуба на контрол върху качеството на отделните подсистеми: невъзможно е да се осигури целевите показатели за качество на една подсистема в откъс от показателите за качество на друга подсистема. Ако подсистемите се разработват от различни екипи, това е голям проблем.

Всичко става още по-интересно, ако интегрираните подсистеми се намират в различни приложения и е необходимо да се направят синхронни промени от двете страни. Как да се осигури транзакционността на тези промени?

Ако промените се въвеждат с отделни транзакции, ще бъде необходимо да се осигури надеждна обработка на изключения и компенсации, а това напълно неутрализира основното предимство на синхронните интеграции – простотата.

Наум идват и разпределените транзакции, но ние не ги използваме в нашите решения: трудно е да се осигури надеждност.

«Сага» като решение на проблема с транзакциите

С растежа на популярността на микросервисите, все по-голямо търсене набира Saga Pattern.

Този шаблон отлично решава горепосочените проблеми на дългите транзакции, а също така разширява възможностите за управление на състоянието на системата от страна на бизнес логиката: компенсацията след неуспешна транзакция може да не върне системата в първоначалното състояние, а да осигури алтернативен маршрут за обработка на данни. Това също така позволява да не се повтарят успешно завършените стъпки при повторни опити да се доведе процесът до «добро» завършване.

Интересно е, че в монолитни системи този шаблон също е актуален, ако става въпрос за интеграция на слабо свързани подсистеми и се наблюдават негативни ефекти, причинени от дълги транзакции и съответни блокировки на ресурсите.

В контекста на нашите бизнес процеси в стил BPM, имплементирането на «Саги» се оказва много лесно: отделните стъпки на «Сагата» могат да бъдат зададени като активности в бизнес процеса, а персистентното състояние на бизнес процеса определя и вътрешното състояние на «Сагата». Тоест, не ни е необходим никакъв допълнителен координационен механизъм. Нужно е само съобщителен брокер с поддръжка на гаранции «поне веднъж» като транспорт.

Но и такова решение има своята «цена»:

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

За монолитни системи оправдаността на използването на 'Saga' не е толкова очевидна. За микросервизи и други SOA, където вероятно вече има брокер, а пълната последователност е била пожертвана от самото начало на проекта, ползата от използването на този шаблон може значително да надвиши недостатъците, особено при наличие на удобен 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()

// Това е клас на инстанция на процес: инкапсулира вътрешното му състояние
class InitialPlayer : ProcessImpl(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(name = "InitialPlayer",
                                                     version = 1) {

    // Според правилата, първият играч е инициатор на играта и трябва да бъде единствен
    uniqueConstraint = UniqueConstraints.singleton

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

private fun MessageSendInstance.selectNextPlayer() {
    val player = process.players.random()
    receiverDomain = player.domain
    receiverProcessInstanceId = player.id
    println("Стъпка ${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()

class RandomPlayer : ProcessImpl(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(name = "RandomPlayer", 
                                                   version = 1) {

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

class RoundRobinPlayer : ProcessImpl(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(
        name = "RoundRobinPlayer", 
        version = 1) {

    val waitNewGameSignal = signalWait("NewGame")
    val waitStopGameSignal = signalWait("StopGame")
    val sendPlayerJoin = signal("PlayerJoin") {
        signalData = { playerName }
    }
    val sendPlayerOut = signal("PlayerOut") {
        signalData = { playerName }
    }
    val waitPlayerJoin = signalWaitCustom("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("PlayerOut") { signal ->
        players.remove(PlayerInfo(
                signal.data!!, 
                signal.sender.domain, 
                signal.sender.processInstanceId))
        allPlayersOut = players.isEmpty()
    }
    val sendHandshake = messageSend("Handshake") {
        messageData = { playerName }
        activation = {
            receiverDomain = process.players.last().domain
            receiverProcessInstanceId = process.players.last().id
        }
    }
    val receiveHandshake = messageWait("Handshake") { message ->
        if (!players.any { message.sender.processInstanceId == it.id}) {
            players.add(PlayerInfo(
                    message.data!!, 
                    message.sender.domain, 
                    message.sender.processInstanceId))
        }
    }
    val throwBall = messageSend("Ball") {
        messageData = { shotCounter + 1 }
        activation = { selectNextPlayer() }
        onEntry { energy -= 1 }
    }
    val waitBall = messageWaitData("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.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 playerParams(String name, int energy) {
    Map params = new HashMap();
    params.put("playerName", name);
    params.put("energy", energy);
    return params;
}

Стартираме теста, гледаме логовете:

конзолен изход

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

От всичко това можем да извлечем няколко важни извода:

при наличието на необходимите инструменти, приложните разработчици могат да създават интеграционни взаимодействия между приложенията, без да се отклоняват от бизнес логиката;
сложността (complexity) на интеграционната задача, изискваща инженерни компетенции, може да бъде скрита в рамките на фреймворка, ако това е първоначално заложено в архитектурата на фреймворка. Трудността на задачата (difficulty) не може да бъде скрита, поради което решаването на трудната задача в кода ще изглежда съответно;
при разработването на интеграционната логика е задължително да се вземат предвид eventually consistency и отсъствието на линеаризируемост на промените в състоянието на всички участници в интеграцията. Това налага усложняване на логиката, за да я направи нечувствителна на реда на възникване на външните събития. В нашия пример играчът трябва да участва в играта едва след като обяви излизането си от играта: другите играчи ще продължат да му предават топката, докато информацията за излизането му не достигне и не бъде обработена от всички участници. Тази логика не произлиза от правилата на играта и е компромисно решение в рамките на избраната архитектура.

След това ще говорим за различните нюанси на нашето решение, компромисите и други моменти.

Всички съобщения – в една опашка

Всички интегрирани приложения работят с една интеграционна шина, представена под формата на външен брокер, една опашка BPMQueue – за съобщения и една тема BPMTopic – за сигнали (събития). Пропускането на всички съобщения през една опашка само по себе си е компромис. На ниво бизнес логика вече могат да бъдат въведени колкото се може нови типове съобщения, без да се внасят промени в структурата на системата. Това е значително опростяване, но носи със себе си определени рискове, които в контекста на нашите типови задачи ни се струват не толкова значителни.

Въпреки това, тук има една особеност: всяко приложение филтрира "своите" съобщения от опашката още на входа, по името на своя домейн. Освен това домейнът може да бъде указан и в сигналите, ако е необходимо да се ограничи "областта на видимост" на сигнала до едно единствено приложение. Това трябва да увеличи пропускната способност на шината, но бизнес логиката сега трябва да работи с имената на домейните: за адресиране на съобщения – задължително, за сигналите – желателно.

Осигуряване на надеждност на интеграционната шина

Надеждността се състои от няколко елемента:

избраният брокер на съобщения – критично важен компонент на архитектурата и единична точка на неуспех: той трябва да бъде достатъчно отказоустойчив. Трябва да се използват само изпитани на времето реализации, с добра поддръжка и голяма общност;
необходимо е да се осигури висока наличност на брокера на съобщения, за което той трябва да бъде физически отделен от интегрираните приложения (осигуряването на висока наличност на приложенията с бизнес логика е значително по-сложно и скъпо);
брокерът е задължен да осигури гаранции за доставка "поне веднъж". Това е задължително изискване за надеждната работа на интеграционната шина. За гаранции с ниво "точно веднъж" няма нужда: бизнес процесите, като правило, не са чувствителни към повторното постъпване на съобщения или събития, а в особени случаи, където това е важно, е по-лесно да се добави допълнителна проверка в бизнес логиката, отколкото постоянно да се използват достатъчно "скъпи" гаранции;
изпращането на съобщения и сигнали трябва да бъде включено в обща транзакция с промяна на състоянието на бизнес процесите и домейните данни. Предпочитаният вариант ще бъде използването на модела Transactional Outbox, но той ще изисква наличието на допълнителна таблица в базата и ретранслятор. В JEE приложенията може да се опрости този момент чрез използването на локален JTA мениджър, но свързването с избрания брокер трябва да може да работи в режим XA;
обработчиците на входящи съобщения и събития също трябва да работят с трансакцията по промяна на състоянието на бизнес процеса: ако такава трансакция се отмени, тогава и приемането на съобщението трябва да бъде отменено;
съобщенията, които не са могли да бъдат доставени поради грешки, трябва да се складират в отделно хранилище DLQ (Dead Letter Queue). Създадохме отделен платформа микросервис за това, който съхранява такива съобщения в своето хранилище, индексира ги по атрибути (за бърза групировка и търсене) и предоставя API за преглед, повторно изпращане до дестинация и изтриване на съобщенията. Администраторите на системата могат да работят с този сервис чрез своя уеб интерфейс;
в настройките на брокера е необходимо да се настрои броят на опитите за доставка и времето за забавяне между доставките, за да се намали вероятността за попадането на съобщения в DLQ (определянето на оптимални параметри практически е невъзможно, но може да се действа емпирично и да се настройват по време на експлоатацията);
хранилището на DLQ трябва непрекъснато да се мониторира, а системата за мониторинг трябва да известява администраторите на системата, за да реагират възможно най-бързо при появата на недоставени съобщения. Това ще намали "зоната на поражение" от възникналия отказ или грешка в бизнес логиката;
интеграционната шина трябва да бъде нечувствителна към временно отсъствие на приложения: абонаментите за темата трябва да бъдат durable, а доменът на приложението трябва да бъде уникален, за да не се опита някой друг да обработи съобщенията от опашката, докато приложението отсъства.

Осигуряване на потокова безопасност на бизнес логиката

На един и същ екземпляр на бизнес процеса могат да постъпят едновременно няколко съобщения и събития, чиято обработка ще стартира паралелно. В същото време за приложния разработчик всичко трябва да бъде просто и потоково безопасно.

Бизнес логиката на процеса обработва всяко външно събитие, което влияе на този бизнес процес, поотделно. Такива събития могат да бъдат:

стартиране на екземпляр на бизнес процес;
действие на потребителя, свързано с активността в бизнес процеса;
постъпване на съобщение или сигнал, на който е абониран екземплярът на бизнес процеса;
сработване на таймер, зададен от екземпляра на бизнес процеса;
управляващо въздействие чрез 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