Dobrý den, Habr!
Naše společnost se specializuje na vývoj softwarových řešení třídy ERP, ve kterých lví podíl zaujímají transakční systémy s obrovským množstvím obchodní logiky a workflow a la EDMS. Moderní verze našich produktů jsou založeny na technologiích JavaEE, ale také aktivně experimentujeme s mikroslužbami. Jednou z nejproblematičtějších oblastí takových řešení je integrace různých subsystémů souvisejících se sousedními doménami. Úkoly integrace nás vždy velmi trápily, bez ohledu na architektonické styly, technologické sady a rámce, které používáme, ale v poslední době došlo v řešení těchto problémů k pokroku.
V článku, na který jste upozornili, budu hovořit o zkušenostech a architektonickém průzkumu NPO Krista v určeném území. Zvážíme také příklad jednoduchého řešení integračního problému z pohledu vývojáře aplikace a zjistíme, co se za touto jednoduchostí skrývá.
Zřeknutí se odpovědnosti
Architektonická a technická řešení popsaná v článku nabízím na základě osobních zkušeností v rámci konkrétních úkolů. Tato řešení netvrdí, že jsou univerzální a za jiných podmínek použití nemusí být optimální.
Co s tím má společného BPM?
Abychom na tuto otázku odpověděli, musíme se trochu ponořit do specifik aplikovaných problémů našich řešení. Hlavní součástí obchodní logiky v našem typickém transakčním systému je zadávání dat do databáze přes uživatelská rozhraní, manuální a automatická kontrola těchto dat, jejich předávání nějakým workflow, publikování do jiného systému / analytické databáze / archivu, generování reportů. Klíčovou funkcí systému pro zákazníky je tedy automatizace jejich interních obchodních procesů.
Pro usnadnění používáme v komunikaci termín „dokument“ jako nějakou abstrakci souboru dat, sjednocený společným klíčem, ke kterému lze „připojit konkrétní pracovní postup“.
Ale co integrační logika? Koneckonců, úkol integrace je generován architekturou systému, která je „rozřezána“ na části NE na žádost zákazníka, ale pod vlivem zcela jiných faktorů:
- pod vlivem Conwayova zákona;
- v důsledku opětovného použití subsystémů dříve vyvinutých pro jiné produkty;
- jak rozhodl architekt, na základě nefunkčních požadavků.
Existuje velké pokušení oddělit logiku integrace od obchodní logiky hlavního pracovního postupu, aby nedošlo ke znečištění obchodní logiky integračními artefakty a aby se vývojář aplikace nemusel ponořit do zvláštností architektonického prostředí systému. Tento přístup má řadu výhod, ale praxe ukazuje jeho neefektivnost:
- řešení integračních problémů obvykle sklouzává k nejjednodušším možnostem v podobě synchronních volání z důvodu omezených rozšiřovacích bodů v implementaci hlavního workflow (více o nedostatcích synchronní integrace níže);
- integrační artefakty stále pronikají do hlavní obchodní logiky, když je vyžadována zpětná vazba od jiného subsystému;
- vývojář aplikace integraci ignoruje a může ji snadno prolomit změnou pracovního postupu;
- systém přestává být z pohledu uživatele jedním celkem, stávají se patrné „švy“ mezi subsystémy, objevují se nadbytečné uživatelské operace, které iniciují přenos dat z jednoho subsystému do druhého.
Dalším přístupem je považovat integrační interakce za nedílnou součást základní obchodní logiky a pracovního postupu. Aby požadavky na dovednosti vývojářů aplikací nerostly do nebes, mělo by se vytváření nových integračních interakcí provádět snadno a přirozeně, s minimálními možnostmi výběru řešení. Je to složitější, než to vypadá: nástroj musí být dostatečně výkonný, aby uživateli poskytl potřebnou rozmanitost možností jeho použití a zároveň se nenechal střelit do nohy. Existuje mnoho otázek, na které by měl inženýr odpovědět v souvislosti s integračními úkoly, ale na které by vývojář aplikací neměl myslet při své každodenní práci: hranice transakcí, konzistence, atomicita, bezpečnost, škálování, distribuce zatížení a zdrojů, směrování, zařazování, propagace a přepínání kontextů atd. Vývojářům aplikací je nutné nabídnout vcelku jednoduché rozhodovací šablony, ve kterých jsou již skryty odpovědi na všechny takové otázky. Tyto vzory by měly být dostatečně bezpečné: obchodní logika se velmi často mění, což zvyšuje riziko zavádění chyb, náklady na chyby by měly zůstat na poměrně nízké úrovni.
Ale přesto, co s tím má společného BPM? Existuje mnoho možností pro implementaci workflow...
V našich řešeních je totiž velmi oblíbená další implementace business procesů – prostřednictvím deklarativního nastavení diagramu přechodu stavu a propojení handlerů s business logikou k přechodům. Přitom stav, který určuje aktuální pozici „dokumentu“ v obchodním procesu, je atributem samotného „dokumentu“.
Takto vypadá proces na začátku projektu
Obliba takové implementace je dána relativní jednoduchostí a rychlostí vytváření lineárních obchodních procesů. Jak se však softwarové systémy stávají složitějšími, automatizovaná část obchodního procesu roste a stává se složitější. Je potřeba dekompozice, opětovné použití částí procesů a také forkovací procesy tak, aby každá větev byla vykonávána paralelně. Za takových podmínek se nástroj stává nepohodlným a diagram přechodu stavu ztrácí svůj informační obsah (integrační interakce se v diagramu vůbec neprojevují).
Takto vypadá proces po několika iteracích objasňování požadavků
Východiskem z této situace byla integrace motoru do některých produktů s nejsložitějšími obchodními procesy. Z krátkodobého hlediska mělo toto řešení určitý úspěch: bylo možné implementovat složité obchodní procesy při zachování poměrně informativního a aktuálního diagramu v notaci. .
Malá část složitého obchodního procesu
Z dlouhodobého hlediska řešení nesplnilo očekávání: vysoká pracnost při vytváření obchodních procesů pomocí vizuálních nástrojů neumožňovala dosáhnout přijatelných ukazatelů produktivity a samotný nástroj se stal jedním z nejvíce neoblíbených mezi vývojáři. Objevily se také stížnosti na vnitřní strukturu motoru, což vedlo ke vzniku mnoha „záplat“ a „berliček“.
Hlavním pozitivním aspektem používání jBPM bylo uvědomění si výhod a škod plynoucích z vlastního trvalého stavu pro instanci obchodního procesu. Viděli jsme také možnost využití procesního přístupu k implementaci komplexních integračních protokolů mezi různými aplikacemi pomocí asynchronních interakcí prostřednictvím signálů a zpráv. Zásadní roli v tom hraje přítomnost přetrvávajícího stavu.
Na základě výše uvedeného můžeme dojít k závěru: Procesní přístup ve stylu BPM nám umožňuje řešit širokou škálu úloh pro automatizaci stále složitějších podnikových procesů, harmonicky do těchto procesů zapadat integrační aktivity a zachovat si schopnost vizuálně zobrazit realizovaný proces ve vhodné notaci.
Nevýhody synchronních volání jako integračního vzoru
Synchronní integrace se týká nejjednoduššího blokovacího volání. Jeden subsystém funguje jako strana serveru a zpřístupňuje rozhraní API požadovanou metodou. Další subsystém funguje jako klientská strana a ve správný čas zavolá s očekáváním výsledku. V závislosti na architektuře systému mohou být klientská a serverová strana hostována buď ve stejné aplikaci a procesu, nebo v různých. Ve druhém případě musíte použít nějakou implementaci RPC a zajistit seřazení parametrů a výsledku volání.
Takový integrační vzor má poměrně velkou sadu nedostatků, ale v praxi je velmi široce používán díky své jednoduchosti. Rychlost implementace uchvátí a nutí vás ji aplikovat znovu a znovu v podmínkách „hoření“ termínů, zapisování řešení do technického dluhu. Stává se však také, že jej nezkušení vývojáři používají nevědomě, jednoduše si neuvědomují negativní důsledky.
Kromě nejzřetelnějšího nárůstu konektivity subsystémů existují méně zřejmé problémy s „rozšiřováním“ a „roztahováním“ transakcí. Pokud obchodní logika provede nějaké změny, pak jsou transakce nepostradatelné a transakce zase uzamknou určité aplikační zdroje ovlivněné těmito změnami. To znamená, že dokud jeden podsystém nebude čekat na odpověď od druhého, nebude schopen dokončit transakci a uvolnit zámky. To výrazně zvyšuje riziko různých účinků:
- odezva systému se ztrácí, uživatelé dlouho čekají na odpovědi na požadavky;
- server obecně přestává reagovat na požadavky uživatelů kvůli přeplněnému fondu vláken: většina vláken „stojí“ na zámku zdroje obsazeného transakcí;
- začnou se objevovat uváznutí: pravděpodobnost jejich výskytu silně závisí na době trvání transakcí, množství obchodní logiky a zámcích zahrnutých v transakci;
- objevují se chyby vypršení časového limitu transakce;
- server „spadne“ na OutOfMemory, pokud úloha vyžaduje zpracování a změnu velkého množství dat a přítomnost synchronních integrací velmi ztěžuje rozdělení zpracování na „lehčí“ transakce.
Z architektonického hlediska vede použití blokování volání během integrace ke ztrátě kontroly kvality jednotlivých subsystémů: není možné dosáhnout cílů kvality pro jeden subsystém izolovaně od ukazatelů kvality pro jiný subsystém. Pokud subsystémy vyvíjejí různé týmy, je to velký problém.
Věci jsou ještě zajímavější, pokud jsou integrované subsystémy v různých aplikacích a je třeba provést synchronní změny na obou stranách. Jak provést tyto změny transakční?
Pokud jsou změny prováděny v samostatných transakcích, pak bude potřeba zajistit robustní zpracování výjimek a kompenzaci, a to zcela eliminuje hlavní výhodu synchronních integrací – jednoduchost.
V úvahu připadají i distribuované transakce, které však v našich řešeních nepoužíváme: je obtížné zajistit spolehlivost.
"Sága" jako řešení problému transakcí
S rostoucí popularitou mikroslužeb roste poptávka po nich .
Tento vzorec dokonale řeší výše uvedené problémy dlouhých transakcí a také rozšiřuje možnosti řízení stavu systému ze strany obchodní logiky: kompenzace po neúspěšné transakci nemusí vrátit systém do původního stavu, ale poskytnout alternativu cesta zpracování dat. Umožňuje vám také neopakovat úspěšně dokončené kroky zpracování dat, když se pokusíte proces dovést do „dobrého“ konce.
Je zajímavé, že v monolitických systémech je tento vzor také relevantní, pokud jde o integraci volně propojených subsystémů a existují negativní efekty způsobené dlouhými transakcemi a odpovídajícími uzamčeními zdrojů.
S ohledem na naše obchodní procesy ve stylu BPM se ukazuje, že implementace Ság je velmi snadná: jednotlivé kroky Ság lze nastavit jako činnosti v rámci obchodního procesu a trvalý stav obchodního procesu mimo jiné určuje , vnitřní stav Ság. To znamená, že nepotřebujeme žádný další koordinační mechanismus. Vše, co potřebujete, je zprostředkovatel zpráv s podporou záruk „alespoň jednou“ jako přepravu.
Ale takové řešení má také svou vlastní „cenu“:
- obchodní logika se stává složitější: musíte vypracovat kompenzaci;
- bude nutné opustit plnou konzistenci, která může být zvláště citlivá u monolitických systémů;
- architektura se trochu zkomplikuje, existuje další potřeba zprostředkovatele zpráv;
- budou zapotřebí další nástroje pro monitorování a správu (i když obecně je to dokonce dobré: kvalita systémových služeb se zvýší).
U monolitických systémů není opodstatnění použití "Sags" tak zřejmé. U mikroslužeb a dalších SOA, kde s největší pravděpodobností již existuje zprostředkovatel a na začátku projektu byla obětována plná konzistence, mohou výhody použití tohoto vzoru výrazně převážit nevýhody, zvláště pokud existuje vhodné API na úroveň obchodní logiky.
Zapouzdření obchodní logiky v mikroslužbách
Když jsme začali experimentovat s mikroslužbami, vyvstala rozumná otázka: kam dát doménovou obchodní logiku ve vztahu ke službě, která zajišťuje perzistenci doménových dat?
Při pohledu na architekturu různých BPMS se může zdát rozumné oddělit obchodní logiku od persistence: vytvořit vrstvu mikroslužeb nezávislých na platformě a doméně, které tvoří prostředí a kontejner pro provádění doménové obchodní logiky, a uspořádat perzistenci doménových dat jako samostatnou vrstva velmi jednoduchých a lehkých mikroslužeb. Obchodní procesy v tomto případě organizují služby perzistentní vrstvy.
Tento přístup má velmi velké plus: můžete zvýšit funkčnost platformy, jak chcete, a „ztloustne“ z toho pouze odpovídající vrstva mikroslužeb platformy. Obchodní procesy z jakékoli domény dostanou okamžitě možnost využívat novou funkcionalitu platformy, jakmile je aktualizována.
Podrobnější studie odhalila významné nedostatky tohoto přístupu:
- služba platformy, která vykonává obchodní logiku mnoha domén najednou, s sebou nese velká rizika jako jediný bod selhání. Časté změny obchodní logiky zvyšují riziko chyb vedoucích k selháním celého systému;
- problémy s výkonem: obchodní logika pracuje se svými daty prostřednictvím úzkého a pomalého rozhraní:
- data budou znovu shromážděna a pumpována přes síťový zásobník;
- doménová služba často vrátí více dat, než vyžaduje obchodní logika ke zpracování, kvůli nedostatečným možnostem parametrizace dotazů na úrovni externího API služby;
- několik nezávislých částí obchodní logiky může opakovaně vyžadovat stejná data pro zpracování (tento problém můžete zmírnit přidáním session beans, které ukládají data do mezipaměti, ale to dále komplikuje architekturu a vytváří problémy s čerstvostí dat a znehodnocením mezipaměti);
- transakční problémy:
- obchodní procesy s trvalým stavem uložené službou platformy jsou nekonzistentní s daty domény a neexistují žádné snadné způsoby, jak tento problém vyřešit;
- přesunutí zámku doménových dat mimo transakci: pokud doménová obchodní logika potřebuje provést změny, po první kontrole správnosti skutečných dat je nutné vyloučit možnost konkurenční změny zpracovávaných dat. Externí blokování dat může pomoci vyřešit problém, ale takové řešení s sebou nese další rizika a snižuje celkovou spolehlivost systému;
- další komplikace při aktualizaci: v některých případech je třeba aktualizovat službu persistence a obchodní logiku synchronně nebo v přísném pořadí.
Nakonec jsem se musel vrátit k základům: zapouzdřit doménová data a doménovou obchodní logiku do jedné mikroslužby. Tento přístup zjednodušuje vnímání mikroslužby jako nedílné součásti systému a nezpůsobuje výše uvedené problémy. Toto také není zdarma:
- Standardizace API je vyžadována pro interakci s obchodní logikou (zejména pro poskytování uživatelských aktivit jako součást obchodních procesů) a službami platformy API; je třeba věnovat větší pozornost změnám API, dopředné a zpětné kompatibilitě;
- je nutné přidat další runtime knihovny, aby bylo zajištěno fungování obchodní logiky jako součásti každé takové mikroslužby, a to dává vzniknout novým požadavkům na takové knihovny: lehkost a minimum tranzitivních závislostí;
- vývojáři obchodní logiky musí sledovat verze knihoven: pokud mikroslužba nebyla dlouho dokončena, bude s největší pravděpodobností obsahovat zastaralou verzi knihoven. To může být neočekávanou překážkou pro přidání nové funkce a může vyžadovat migraci staré obchodní logiky takové služby do nových verzí knihoven, pokud mezi verzemi došlo k nekompatibilním změnám.
V takové architektuře je přítomna i vrstva platformových služeb, ale tato vrstva již netvoří kontejner pro vykonávání doménové obchodní logiky, ale pouze její prostředí, poskytující pomocné „platformové“ funkce. Taková vrstva je potřebná nejen pro zachování lehkosti doménových mikroslužeb, ale také pro centralizaci správy.
Například aktivity uživatelů v obchodních procesech generují úkoly. Při práci s úlohami však uživatel musí vidět úlohy ze všech domén v obecném seznamu, což znamená, že musí existovat vhodná služba platformy pro registraci úloh, zbavená obchodní logiky domény. Udržet zapouzdření obchodní logiky v tomto kontextu je poměrně problematické a jde o další kompromis této architektury.
Integrace obchodních procesů očima vývojáře aplikací
Jak již bylo zmíněno výše, vývojář aplikace musí být abstrahován od technických a inženýrských vlastností implementace interakce několika aplikací, aby mohl počítat s dobrou produktivitou vývoje.
Pokusme se vyřešit poměrně obtížný integrační problém, speciálně vymyšlený pro tento článek. Půjde o „herní“ úlohu zahrnující tři aplikace, kde každá z nich definuje nějaké doménové jméno: „app1“, „app2“, „app3“.
Uvnitř každé aplikace jsou spuštěny obchodní procesy, které začnou „hrát míč“ prostřednictvím integrační sběrnice. Zprávy pojmenované "Ball" budou fungovat jako míč.
Pravidla hry:
- první hráč je iniciátor. Pozve do hry další hráče, hru zahájí a může ji kdykoli ukončit;
- ostatní hráči deklarují účast ve hře, „seznámí se“ mezi sebou a prvním hráčem;
- po obdržení míče si hráč vybere dalšího zúčastněného hráče a přihraje mu míč. Počítá se celkový počet průchodů;
- každý hráč má "energii", která klesá s každým přihráním míče tímto hráčem. Když energie dojde, hráč je vyřazen ze hry a oznámí jeho odchod do důchodu;
- pokud hráč zůstane sám, okamžitě ohlásí svůj odchod;
- když jsou všichni hráči vyřazeni, první hráč prohlásí konec hry. Pokud opustil hru dříve, pak zbývá sledovat hru, aby ji dokončil.
K vyřešení tohoto problému použiji naše DSL pro obchodní procesy, které vám umožní popsat logiku v Kotlinu kompaktně, s minimem kotlinového materiálu.
V aplikaci app1 bude fungovat obchodní proces prvního hráče (je také iniciátorem hry):
třídy 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}")
}Kromě provádění obchodní logiky může výše uvedený kód vytvořit objektový model obchodního procesu, který lze vizualizovat jako diagram. Ještě jsme neimplementovali vizualizér, takže jsme museli strávit nějaký čas kreslením (zde jsem mírně zjednodušil zápis BPMN ohledně použití hradel, abych zlepšil konzistenci diagramu s výše uvedeným kódem):
app2 bude zahrnovat obchodní proces jiného hráče:
třída 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}")
}Diagram:
V aplikaci app3 uděláme z hráče trochu jiné chování: místo náhodného výběru dalšího hráče bude jednat podle algoritmu round-robin:
třídy 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}")
}Jinak se chování hráče od předchozího neliší, takže se diagram nemění.
Nyní potřebujeme test, abychom to všechno mohli spustit. Uvedu pouze kód samotného testu, abych článek nezahltil standardním vzorem (ve skutečnosti jsem použil dříve vytvořené testovací prostředí k testování integrace dalších obchodních procesů):
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;
}Spusťte test, podívejte se na protokol:
výstup konzoly
Взята блокировка ключа 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!Z toho všeho lze vyvodit několik důležitých závěrů:
- jsou-li k dispozici potřebné nástroje, mohou vývojáři aplikací vytvářet integrační interakce mezi aplikacemi, aniž by se odtrhli od obchodní logiky;
- složitost (složitost) integračního úkolu, který vyžaduje inženýrské kompetence, může být skryta uvnitř rámce, pokud je původně stanovena v architektuře rámce. Obtížnost úkolu (obtížnost) nelze skrýt, takže podle toho bude vypadat řešení obtížného úkolu v kódu;
- při rozvíjení integrační logiky je nutné vzít v úvahu případnou konzistenci a nelinearizovatelnost změny stavu všech účastníků integrace. To nás nutí komplikovat logiku, aby byla necitlivá na pořadí, ve kterém se vnější události odehrávají. V našem příkladu je hráč nucen zúčastnit se hry poté, co oznámí svůj odchod ze hry: ostatní hráči mu budou nadále podávat míč, dokud se informace o jeho odchodu nedostane a není zpracována všemi účastníky. Tato logika nevyplývá z pravidel hry a je kompromisním řešením v rámci zvolené architektury.
Dále si promluvme o různých jemnostech našeho řešení, kompromisech a dalších bodech.
Všechny zprávy v jedné frontě
Všechny integrované aplikace pracují s jednou integrační sběrnicí, která je reprezentována jako externí broker, jedním BPMQueue pro zprávy a jedním tématem BPMTopic pro signály (události). Předávání všech zpráv jedinou frontou je samo o sobě kompromisem. Na úrovni obchodní logiky nyní můžete zavádět tolik nových typů zpráv, kolik chcete, aniž byste museli měnit strukturu systému. Jde o výrazné zjednodušení, které však nese určitá rizika, která se nám v kontextu našich typických úkolů zdála ne tak významná.
Je zde však jedna jemnost: každá aplikace filtruje „své“ zprávy z fronty u vstupu podle názvu své domény. V signálech lze také specifikovat doménu, pokud potřebujete omezit „rozsah“ signálu na jednu jedinou aplikaci. To by mělo zvýšit šířku pásma sběrnice, ale obchodní logika nyní musí pracovat s názvy domén: povinné pro adresování zpráv, žádoucí pro signály.
Zajištění spolehlivosti integrační sběrnice
Spolehlivost se skládá z několika věcí:
- Zvolený zprostředkovatel zpráv je kritickou součástí architektury a jediným bodem selhání: musí být dostatečně odolný proti chybám. Měli byste používat pouze časem prověřené implementace s dobrou podporou a velkou komunitou;
- je nutné zajistit vysokou dostupnost zprostředkovatele zpráv, pro který musí být fyzicky oddělen od integrovaných aplikací (vysoká dostupnost aplikací s aplikovanou business logikou je mnohem obtížnější a nákladnější na zajištění);
- zprostředkovatel je povinen poskytnout záruky dodání „alespoň jednou“. To je povinný požadavek pro spolehlivý provoz integrační sběrnice. Záruky úrovně „přesně jednou“ nejsou potřeba: obchodní procesy obvykle nejsou citlivé na opakovaný příchod zpráv nebo událostí a ve speciálních úkolech, kde je to důležité, je snazší přidat další kontroly do obchodní logiky, než neustále používat spíše "drahé" " záruky;
- odesílání zpráv a signálů musí být součástí společné transakce se změnou stavu obchodních procesů a doménových dat. Preferovanou možností by bylo použít vzor , ale bude to vyžadovat další tabulku v databázi a přenos. V aplikacích JEE to lze zjednodušit použitím místního správce JTA, ale připojení k vybranému brokerovi musí fungovat v režimu ;
- manipulátory příchozích zpráv a událostí musí také pracovat s transakcí změny stavu obchodního procesu: pokud je taková transakce vrácena zpět, musí být také zrušen příjem zprávy;
- zprávy, které nemohly být doručeny kvůli chybám, by měly být uloženy v samostatném obchodě (Fronta na mrtvé dopisy). Za tímto účelem jsme vytvořili samostatnou mikroslužbu platformy, která takové zprávy ukládá do svého úložiště, indexuje je podle atributů (pro rychlé seskupování a vyhledávání) a zpřístupňuje API pro prohlížení, opětovné odesílání na cílovou adresu a mazání zpráv. Správci systému mohou s touto službou pracovat prostřednictvím svého webového rozhraní;
- v nastavení brokera je potřeba upravit počet opakování doručení a zpoždění mezi doručením, aby se snížila pravděpodobnost, že se zprávy dostanou do DLQ (je téměř nemožné vypočítat optimální parametry, ale můžete jednat empiricky a upravit je během úkon);
- úložiště DLQ by mělo být nepřetržitě monitorováno a monitorovací systém by měl upozornit správce systému, aby mohli co nejrychleji reagovat, když se objeví nedoručené zprávy. Tím se sníží „poškozená zóna“ selhání nebo chyby obchodní logiky;
- integrační sběrnice musí být necitlivá na dočasnou absenci aplikací: předplatné témat musí být trvalé a název domény aplikace musí být jedinečný, aby se někdo jiný nepokusil zpracovat její zprávu z fronty během nepřítomnosti aplikace.
Zajištění bezpečnosti vláken obchodní logiky
Stejná instance obchodního procesu může přijímat několik zpráv a událostí najednou, jejichž zpracování začne paralelně. Zároveň by pro vývojáře aplikací mělo být vše jednoduché a bezpečné pro vlákna.
Obchodní logika procesů zpracovává každou externí událost, která ovlivňuje tento obchodní proces, individuálně. Tyto události mohou být:
- spuštění instance obchodního procesu;
- akce uživatele související s aktivitou v rámci obchodního procesu;
- přijetí zprávy nebo signálu, ke kterému je přihlášena instance obchodního procesu;
- vypršení časovače nastaveného instancí obchodního procesu;
- ovládací akce přes API (např. přerušení procesu).
Každá taková událost může změnit stav instance obchodního procesu: některé aktivity mohou skončit a jiné začít, hodnoty trvalých vlastností se mohou změnit. Uzavření jakékoli aktivity může mít za následek aktivaci jedné nebo více z následujících aktivit. Ti zase mohou přestat čekat na další události, nebo pokud nepotřebují žádná další data, mohou dokončit stejnou transakci. Před uzavřením transakce se nový stav obchodního procesu uloží do databáze, kde bude čekat na další externí událost.
Trvalá data obchodních procesů uložená v relační databázi jsou velmi pohodlným bodem synchronizace zpracování při použití SELECT FOR UPDATE. Pokud se jedné transakci podařilo získat stav obchodního procesu ze základny, aby jej změnil, pak žádná další transakce paralelně nebude schopna získat stejný stav pro další změnu a po dokončení první transakce je druhá transakce zaručeno, že obdrží již změněný stav.
Pomocí pesimistických zámků na straně DBMS splňujeme všechny potřebné požadavky a také si zachovat možnost škálovat aplikaci obchodní logiky zvýšením počtu spuštěných instancí.
Pesimistické zámky nás však ohrožují uváznutím, což znamená, že SELECT FOR UPDATE by měl být stále omezen na nějaký rozumný časový limit v případě zablokování v některých závažných případech v obchodní logice.
Dalším problémem je synchronizace startu obchodního procesu. I když neexistuje žádná instance obchodního procesu, není ani v databázi žádný stav, takže popsaná metoda nebude fungovat. Pokud chcete zajistit jedinečnost instance obchodního procesu v určitém rozsahu, potřebujete nějaký druh synchronizačního objektu přidruženého ke třídě procesu a odpovídajícímu rozsahu. K vyřešení tohoto problému používáme jiný zamykací mechanismus, který nám umožňuje zamknout libovolný zdroj určený klíčem ve formátu URI prostřednictvím externí služby.
V našich příkladech obchodní proces InitialPlayer obsahuje prohlášení
uniqueConstraint = UniqueConstraints.singletonProto protokol obsahuje zprávy o odebrání a uvolnění zámku odpovídajícího klíče. Pro jiné obchodní procesy takové zprávy neexistují: uniqueConstraint není nastaven.
Problémy obchodních procesů s přetrvávajícím stavem
Někdy trvalý stav nejen pomáhá, ale také skutečně brzdí vývoj.
Problémy začínají, když potřebujete provést změny v obchodní logice a/nebo modelu obchodních procesů. Žádná taková změna není shledána jako kompatibilní se starým stavem obchodních procesů. Pokud je v databázi mnoho „živých“ instancí, pak provádění nekompatibilních změn může způsobit spoustu problémů, se kterými jsme se často setkávali při používání jBPM.
V závislosti na hloubce změny můžete jednat dvěma způsoby:
- vytvořte nový typ obchodního procesu, abyste neprováděli nekompatibilní změny starého, a použijte jej místo starého při spouštění nových instancí. Staré instance budou nadále fungovat „starým způsobem“;
- migrovat trvalý stav obchodních procesů při aktualizaci obchodní logiky.
První způsob je jednodušší, ale má svá omezení a nevýhody, například:
- duplikace obchodní logiky v mnoha modelech obchodních procesů, zvýšení objemu obchodní logiky;
- často je vyžadován okamžitý přechod na novou obchodní logiku (téměř vždy z hlediska integračních úkolů);
- vývojář neví, v jakém okamžiku je možné vymazat zastaralé modely.
V praxi používáme oba přístupy, ale učinili jsme řadu rozhodnutí, jak si zjednodušit život:
- v databázi je trvalý stav obchodního procesu uložen ve snadno čitelné a snadno zpracovatelné podobě: v řetězci formátu JSON. To vám umožní provádět migrace uvnitř aplikace i mimo ni. V extrémních případech jej můžete také vyladit pomocí úchytů (obzvláště užitečné při vývoji během ladění);
- integrační obchodní logika nepoužívá názvy obchodních procesů, takže kdykoli je možné nahradit implementaci jednoho ze zúčastněných procesů novým, s novým názvem (například „InitialPlayerV2“). K vazbě dochází prostřednictvím názvů zpráv a signálů;
- model procesu má číslo verze, které zvýšíme, pokud v tomto modelu provedeme nekompatibilní změny, a toto číslo se uloží spolu se stavem instance procesu;
- trvalý stav procesu je nejprve načten ze základny do vhodného objektového modelu, se kterým může migrační procedura pracovat, pokud se změnilo číslo verze modelu;
- postup migrace je umístěn vedle obchodní logiky a nazývá se „líný“ pro každou instanci obchodního procesu v době jeho obnovení z databáze;
- pokud potřebujete rychle a synchronně migrovat stav všech instancí procesů, používají se klasičtější řešení migrace databází, ale tam musíte pracovat s JSON.
Potřebuji další rámec pro obchodní procesy?
Řešení popsaná v článku nám umožnila výrazně zjednodušit život, rozšířit okruh problémů řešených na úrovni vývoje aplikací a zatraktivnit myšlenku oddělení obchodní logiky do mikroslužeb. Za tímto účelem bylo vykonáno mnoho práce, byl vytvořen velmi „odlehčený“ rámec pro podnikové procesy a také komponenty služeb pro řešení identifikovaných problémů v kontextu široké škály aplikovaných úloh. Chceme se o tyto výsledky podělit, převést vývoj společných komponent do otevřeného přístupu pod svobodnou licencí. To bude vyžadovat určité úsilí a čas. Pochopení poptávky po takových řešeních by pro nás mohlo být další pobídkou. V navrhovaném článku je věnována velmi malá pozornost schopnostem samotného rámce, ale některé z nich jsou patrné z uvedených příkladů. Pokud náš framework přesto zveřejníme, bude mu věnován samostatný článek. Mezitím budeme vděční, když nám zanecháte malou zpětnou vazbu odpovědí na otázku:
Průzkumu se mohou zúčastnit pouze registrovaní uživatelé. , prosím.
Potřebuji další rámec pro obchodní procesy?
-
18,8%Ano, něco takového jsem dlouho hledal.
-
12,5%je zajímavé dozvědět se více o vaší implementaci, může to být užitečné2
-
6,2%používáme jeden ze stávajících frameworků, ale uvažujeme o jeho nahrazení1
-
18,8%používáme jeden ze stávajících frameworků, vše vyhovuje3
-
18,8%zvládání bez rámce 3
-
25,0%napište vlastní 4
Hlasovalo 16 uživatelů. 7 uživatelů se zdrželo hlasování.
Zdroj: www.habr.com