Integration im BPM-Stil

Hallo, Habr!

Unser Unternehmen ist auf die Entwicklung von ERP-Softwarelösungen spezialisiert, wobei ein wesentlicher Teil transaktionale Systeme mit umfangreicher Geschäftslogik und Dokumentenverkehr im Stil der elektronischen Dokumentenverwaltung ausmacht. Die modernen Versionen unserer Produkte basieren auf JavaEE-Technologien, jedoch experimentieren wir auch aktiv mit Mikrodiensten. Eines der problematischsten Bereiche solcher Lösungen ist die Integration verschiedener Teilsysteme, die zu angrenzenden Domänen gehören. Die Integrationsaufgaben haben uns immer große Kopfschmerzen bereitet, unabhängig von den von uns verwendeten Architekturmustern, Technologie-Stacks und Frameworks, jedoch gab es in letzter Zeit Fortschritte bei der Lösung solcher Aufgaben.

In dem vorliegenden Artikel werde ich über die Erfahrungen und architektonischen Überlegungen der NPO "Krista" auf diesem Gebiet berichten. Außerdem werden wir ein einfaches Beispiel für eine Integrationsaufgabe aus der Sicht eines Anwendungsentwicklers betrachten und herausfinden, was hinter dieser Einfachheit steckt.

Haftungsausschluss

Die in diesem Artikel beschriebenen architektonischen und technischen Lösungen basieren auf meiner persönlichen Erfahrung im Kontext spezifischer Aufgaben. Diese Lösungen erheben keinen Anspruch auf Universalisierung und können unter anderen Bedingungen suboptimal sein.

Was hat BPM damit zu tun?

Um diese Frage zu beantworten, müssen wir etwas tiefer in die spezifischen Anwendungsaufgaben unserer Lösungen eintauchen. Der Hauptteil der Geschäftslogik in unserem typischen Transaktionssystem besteht aus der Eingabe von Daten in die Datenbank über Benutzeroberflächen, der manuellen und automatisierten Prüfung dieser Daten, der Durchführung in einem bestimmten Workflow, der Veröffentlichung in ein anderes System / eine Analysedatenbank / ein Archiv sowie der Erstellung von Berichten. Somit ist die Schlüsselrolle des Systems für die Kunden die Automatisierung ihrer internen Geschäftsprozesse.

Zur Vereinfachung verwenden wir in der Kommunikation den Begriff "Dokument" als eine gewisse Abstraktion einer Datensammlung, die durch einen gemeinsamen Schlüssel verbunden ist und an die ein bestimmter Workflow "gebunden" werden kann.
Aber wie steht es um die Integrationslogik? Denn die Aufgabe der Integration wird durch die Architektur des Systems hervorgerufen, die nicht auf Kundenanforderung, sondern durch ganz andere Faktoren "zerstückelt" ist:

• Unter dem Einfluss des Conway-Gesetzes;
• als Ergebnis der Wiederverwendung von Subsystemen, die zuvor für andere Produkte entwickelt wurden;
• auf Beschluss des Architekten, basierend auf nicht-funktionalen Anforderungen.

Es gibt eine große Versuchung, die Integrationslogik von der Geschäftslogik des Haupt-Workflows zu trennen, um die Geschäftslogik nicht mit Integrationsartefakten zu belasten und den Anwendungsentwickler von der Notwendigkeit zu befreien, sich mit den Besonderheiten der architektonischen Landschaft des Systems auseinanderzusetzen. Dieser Ansatz hat einige Vorteile, doch die Praxis zeigt, dass er ineffektiv ist:

• Die Lösung von Integrationsproblemen führt in der Regel zu den einfachsten Varianten in Form von synchronen Aufrufen aufgrund der begrenzten Erweiterungspunkte in der Umsetzung des Haupt-Workflows (zu den Nachteilen der synchronen Integration – weiter unten);
• Integrationsartefakte dringen dennoch in die Hauptgeschäftslogik ein, wenn Rückmeldungen aus einem anderen Subsystem erforderlich sind;
• Der Anwendungsentwickler ignoriert die Integration und kann diese leicht brechen, indem er den Workflow verändert.
• Das System hört auf, aus Sicht des Benutzers ein Gesamtes zu sein, und die "Nähte" zwischen den Teilsystemen werden sichtbar, während überflüssige Benutzeroperationen zur Initiierung des Datenaustauschs zwischen den Teilsystemen auftreten.

Ein alternativer Ansatz besteht darin, Integrationsinteraktionen als wesentlichen Bestandteil der grundlegenden Geschäftslogik und des Workflows zu betrachten. Damit die Anforderungen an die Qualifikation der Anwendungsentwickler nicht ins Unermessliche steigen, sollte die Erstellung neuer Integrationsinteraktionen einfach und mühelos erfolgen, mit minimalen Möglichkeiten zur Wahl der Lösungsansätze. Dies zu erreichen ist komplizierter als es scheint: Das Werkzeug muss leistungsfähig genug sein, um dem Benutzer eine ausreichende Auswahl an Anwendungsmöglichkeiten zu bieten, und gleichzeitig verhindern, dass man sich selbst in Schwierigkeiten bringt. Es gibt viele Fragen, die Ingenieure im Kontext von Integrationsaufgaben beantworten müssen, über die Anwendungsentwickler in ihrem Tagesgeschäft nicht nachdenken sollten: Transaktionsgrenzen, Konsistenz, Atomarität, Sicherheit, Skalierbarkeit, Lastverteilung und -ressourcen, Routing, Marshalling, Verbreitung und Kontextwechsel usw. Anwendungsentwicklern sollten ausreichend einfache Lösungsmuster angeboten werden, die bereits Antworten auf all diese Fragen beinhalten. Diese Muster sollten ausreichend sicher sein: Die Geschäftslogik ändert sich sehr häufig, was die Risiken von Fehlern erhöht, und die Kosten von Fehlern sollten auf einem ausreichend niedrigen Niveau bleiben.

Aber was hat BPM damit zu tun? Es gibt viele Möglichkeiten zur Umsetzung von Workflows...
In der Tat erfreut sich eine andere Form der Geschäftsprozessimplementierung in unseren Lösungen großer Beliebtheit – durch die deklarative Vorgabe von Zustandsübergangsdiagrammen und die Anbindung von Handlern mit Geschäftslogik für die Übergänge. Dabei ist der Zustand, der die aktuelle Position des „Dokuments“ im Geschäftsprozess bestimmt, ein Attribut des „Dokuments“ selbst.

So sieht der Prozess zu Beginn des Projekts aus.

Die Beliebtheit dieser Umsetzung beruht auf der relativen Einfachheit und der Geschwindigkeit der Erstellung linearer Geschäftsprozesse. Mit der ständigen Komplexität von Softwaresystemen wächst jedoch der automatisierte Teil des Geschäftsprozesses und wird komplizierter. Es entsteht die Notwendigkeit zur Dekomposition, zur Wiederverwendung von Prozessteilen sowie zur Verzweigung von Prozessen, sodass jeder Zweig parallel ausgeführt wird. In einem solchen Umfeld wird das Werkzeug unpraktisch, und das Zustandsübergangsdiagramm verliert an Aussagekraft (Integrationsinteraktionen werden auf dem Diagramm überhaupt nicht angezeigt).

So sieht der Prozess nach mehreren Iterationen der Anforderungsdefinition aus.

Die Lösung bestand in der Integration der Engine jBPM in einige Produkte mit den komplexesten Geschäftsprozessen. Kurzfristig war diese Lösung durchaus erfolgreich: Es wurde möglich, komplexe Geschäftsprozesse umzusetzen und gleichzeitig ein informatives und aktuelles Diagramm in der Notation BPMN2.

einer kleinen Teil des komplexen Geschäftsprozesses.

Langfristig hat die Lösung jedoch die Erwartungen nicht erfüllt: Der hohe Aufwand für die Erstellung von Geschäftsprozessen mithilfe visueller Werkzeuge ermöglichte es nicht, akzeptable Produktivitätswerte zu erreichen, und das Tool selbst wurde eines der unbeliebtesten bei den Entwicklern. Auch die interne Struktur der Engine wurde kritisiert, was zur Entstehung vieler „Patchwork-Lösungen“ und „Behelfslösungen“ führte.

Der Hauptvorteil der Anwendung von jBPM war das Bewusstsein für den Nutzen und die Nachteile eines eigenen persistenten Zustands für eine Instanz eines Geschäftsprozesses. Zudem haben wir die Möglichkeit einer prozessorientierten Herangehensweise zur Implementierung komplexer Integrationsprotokolle zwischen verschiedenen Anwendungen festgestellt, wobei asynchrone Interaktionen durch Signale und Nachrichten genutzt werden. Der vorhandene persistente Zustand spielt dabei eine entscheidende Rolle.

Zusammenfassend lässt sich sagen: Der prozessorientierte Ansatz im BPM-Stil ermöglicht es uns, ein breites Spektrum an Aufgaben zur Automatisierung zunehmend komplexer werdender Geschäftsprozesse zu lösen, Integrationsaktivitäten harmonisch in diese Prozesse einzufügen und die Möglichkeit der visuellen Darstellung des realisierten Prozesses in einer dafür geeigneten Notation zu bewahren.

Nachteile synchroner Aufrufe als Integrationsmuster

Unter synchroner Integration versteht man einen einfachsten blockierenden Aufruf. Ein Subsystem fungiert als Serverseite und stellt eine API mit der erforderlichen Methode bereit. Das andere Subsystem fungiert als Clientseite und führt zur richtigen Zeit den Aufruf mit der Erwartung des Ergebnisses aus. Je nach Architektur des Systems können die Client- und Serverseiten entweder in einer Anwendung und einem Prozess oder in verschiedenen plaziert werden. Im zweiten Fall ist es erforderlich, eine bestimmte RPC-Implementierung anzuwenden und das Marshalling von Parametern und Ergebnissen des Aufrufs sicherzustellen.

Ein solches Integrationsmuster weist eine Vielzahl von Nachteilen auf, wird aber aufgrund seiner Einfachheit in der Praxis sehr häufig verwendet. Die Geschwindigkeit der Implementierung ist verlockend und führt dazu, dass es immer wieder unter 'dringenden' Fristen eingesetzt wird, wobei die Lösung als technische Schuld verbucht wird. Es kommt jedoch auch vor, dass unerfahrene Entwickler es unbewusst anwenden, ohne sich über die negativen Folgen im Klaren zu sein.

Neben den offensichtlichsten Verbesserungen der Kohärenz von Teilsystemen gibt es auch weniger offensichtliche Probleme mit dem "Schneiden" und "Dehnen" von Transaktionen. Tatsächlich, wenn die Geschäftslogik Änderungen vornimmt, können Transaktionen nicht vermieden werden, und Transaktionen blockieren ihrerseits bestimmte Ressourcen der Anwendung, die von diesen Änderungen betroffen sind. Das heißt, solange ein Teilsystem nicht auf eine Antwort von einem anderen wartet, kann es die Transaktion nicht abschließen und die Sperren aufheben. Dies erhöht das Risiko von verschiedenen Effekten erheblich:

• die Reaktionsfähigkeit des Systems leidet, Benutzer warten lange auf Antworten auf ihre Anfragen;
• der Server antwortet überhaupt nicht mehr auf Benutzeranfragen wegen eines überfüllten Threadpools: die meisten Threads sind durch die Sperrung der von der Transaktion genutzten Ressource blockiert;
• Deadlocks treten auf: die Wahrscheinlichkeit ihres Auftretens hängt stark von der Dauer der Transaktionen, der Menge der in die Transaktion involvierten Geschäftslogik und den Sperren ab;
• Timeout-Fehler der Transaktionen treten auf;
• Der Server „stürzt“ aufgrund von OutOfMemory ab, wenn eine Aufgabe die Verarbeitung und Änderung großer Datenmengen erfordert, und die Verwendung von synchronen Integrationen erschwert das Aufteilen der Verarbeitung in leichtere Transaktionen erheblich.

Aus architektonischer Sicht führt die Verwendung von blockierenden Aufrufen bei der Integration zu einem Verlust der Kontrolle über die Qualität einzelner Teilsysteme: Es ist unmöglich, die Qualitätsziele eines Teilsystems unabhängig von den Qualitätskennzahlen eines anderen Teilsystems zu erfüllen. Wenn die Teilsysteme von verschiedenen Teams entwickelt werden, wird dies zu einem großen Problem.

Es wird noch interessanter, wenn die integrierten Teilsysteme in verschiedenen Anwendungen vorhanden sind und synchron Änderungen von beiden Seiten erforderlich sind. Wie lässt sich die Transaktionalität dieser Änderungen gewährleisten?

Wenn Änderungen in separaten Transaktionen vorgenommen werden, ist es notwendig, eine zuverlässige Fehlerbehandlung und Kompensation zu gewährleisten, was den Hauptvorteil synchroner Integrationen – die Einfachheit – vollständig neutralisiert.

Auch verteilte Transaktionen kommen in den Sinn, aber wir verwenden sie nicht in unseren Lösungen: Es ist schwierig, Zuverlässigkeit zu gewährleisten.

Das Saga-Pattern als Lösung für das Transaktionsproblem

Mit der wachsenden Beliebtheit von Microservices gewinnt Saga-Pattern.

Dieses Pattern löst die zuvor genannten Probleme langer Transaktionen hervorragend und erweitert die Möglichkeiten des Statusmanagements auf Seiten der Geschäftslogik: Eine Kompensation nach einer fehlgeschlagenen Transaktion kann das System nicht in seinen Ursprungszustand zurückversetzen, sondern einen alternativen Datenverarbeitungsweg bereitstellen. Dadurch wird auch vermieden, erfolgreich abgeschlossene Verarbeitungsschritte bei erneuten Versuchen zur Erreichung eines 'guten' Abschlusses zu wiederholen.

Interessanterweise ist dieses Pattern auch in monolithischen Systemen relevant, wenn es um die Integration schwach gekoppelt Subsysteme geht und negative Effekte beobachtet werden, die durch lange Transaktionen und die damit verbundenen Ressourcenblockaden verursacht werden.

In Bezug auf unsere BPM-ähnlichen Geschäftsprozesse ist die Implementierung von „Sagas“ sehr einfach: Die einzelnen Schritte einer „Saga“ können als Aktivitäten innerhalb des Geschäftsprozesses definiert werden, und der persistente Zustand des Geschäftsprozesses bestimmt auch den internen Zustand der „Saga“. Das bedeutet, dass wir keinen zusätzlichen Koordinierungsmechanismus benötigen. Es wird lediglich ein Nachrichtenbroker mit Unterstützung für „at least once“ Garantien als Transport benötigt.

Aber auch diese Lösung hat ihren „Preis“:

• Die Geschäftslogik wird komplexer: Rückerstattungen müssen berücksichtigt werden;
• Es muss auf full consistency verzichtet werden, was besonders problematisch für monolithische Systeme sein kann;
• Die Architektur wird etwas komplizierter, es besteht zusätzlich Bedarf an einem Nachrichtenbroker;
• Zusätzliche Überwachungs- und Verwaltungstools werden erforderlich sein (obwohl das insgesamt sogar positiv ist: Die Servicequalität des Systems wird steigen).

Für monolithische Systeme ist die Rechtfertigung für die Nutzung von "Sagas" nicht so offensichtlich. Bei Microservices und anderen SOA, in denen wahrscheinlich bereits ein Broker vorhanden ist und die vollständige Konsistenz bereits zu Beginn des Projekts geopfert wurde, kann der Nutzen der Verwendung dieses Musters die Nachteile erheblich überwiegen, insbesondere wenn eine benutzerfreundliche API auf der Ebene der Geschäftslogik vorhanden ist.

Kapselung der Geschäftslogik in Microservices

Als wir begannen, mit Microservices zu experimentieren, stellte sich die berechtigte Frage: Wo platziert man die domänenspezifische Geschäftslogik in Bezug auf den Dienst, der die Persistenz der domänenspezifischen Daten gewährleistet?

Wenn man sich die Architektur verschiedener BPMS ansieht, scheint es sinnvoll, die Geschäftslogik von der Persistenz zu trennen: Ein Layer von plattformunabhängigen und domänenunabhängigen Microservices zu schaffen, die eine Umgebung und einen Container für die Ausführung der domänenspezifischen Geschäftslogik bilden, während die Persistenz der domänenspezifischen Daten in einem separaten Layer aus sehr einfachen und leichten Microservices realisiert wird. In diesem Fall orchestrieren die Geschäftsprozesse die Dienste des Persistenz-Layers.

Dieser Ansatz hat einen großen Vorteil: Die Funktionalität der Plattform kann beliebig erweitert werden, und nur die entsprechenden Schichten der Plattform-Mikroservices werden dadurch „dicker“. Geschäftsprozesse aus allen Domänen erhalten sofort die Möglichkeit, neue Funktionen der Plattform zu nutzen, sobald sie aktualisiert werden.

Eine detailliertere Analyse hat wesentliche Nachteile dieses Ansatzes aufgezeigt:

• Der plattformbasierte Dienst, der die Geschäftslogik für mehrere Domänen ausführt, birgt große Risiken als single point of failure. Häufige Änderungen der Geschäftslogik erhöhen das Risiko von Fehlern, die zu Systemausfällen führen, die sich auf das gesamte System ausbreiten;
• Leistungsprobleme: Die Geschäftslogik arbeitet über eine enge und langsame Schnittstelle mit ihren Daten:
  • Daten müssen unnötig oft marshalliert und über den Netzwerk-Stack übertragen werden;
  • Der Domänendienst gibt oft mehr Daten zurück, als die Geschäftslogik für die Verarbeitung benötigt, da die Möglichkeiten zur Parameterisierung von Anfragen auf der Ebene der externen API des Dienstes unzureichend sind;
  • Mehrere unabhängige Teile der Geschäftslogik können dieselben Daten zur Verarbeitung erneut anfordern (dieses Problem kann gemildert werden, indem Sitzungs-Komponenten hinzugefügt werden, die Daten cachen, was jedoch die Architektur zusätzlich erschwert und Probleme mit der Aktualität der Daten und der Cache-Invalidierung schafft);
• Transaktionsprobleme:
  • Geschäftsprozesse mit persistentem Zustand, dessen Speicherung durch den Plattformdienst erfolgt, können von den Domain-Daten abweichen, und einfache Lösungen für dieses Problem sind nicht in Sicht;
  • Auslagerung der Sperrung von Domain-Daten außerhalb der Transaktion: Wenn die Domain-Geschäftslogik Änderungen vornehmen muss, nachdem die Gültigkeit der aktuellen Daten überprüft wurde, muss die Möglichkeit eines konkurrierenden Datenänderungsprozesses ausgeschlossen werden. Eine externe Daten-Sperrung kann helfen, diese Aufgabe zu lösen, bringt jedoch zusätzliche Risiken mit sich und beeinträchtigt die Gesamtreliabilität des Systems;
• Zusätzliche Komplikationen bei der Aktualisierung: In einigen Fällen müssen der Persistenzdienst und die Geschäftslogik synchron oder in strenger Reihenfolge aktualisiert werden.

Letztendlich mussten wir zu den Grundlagen zurückkehren: die Domain-Daten und die Domain-Geschäftslogik in einen einzigen Microservice zu kapseln. Dieser Ansatz erleichtert die Wahrnehmung des Microservice als ganzheitlichen Bestandteil des Systems und vermeidet die oben genannten Probleme. Dies ist jedoch nicht ohne Kosten verbunden:

• es wird eine Standardisierung der APIs für die Interaktion mit der Geschäftslogik benötigt (insbesondere zur Unterstützung der Benutzeraktivitäten im Rahmen der Geschäftsprozesse) und der API-Plattformdienste; es erfordert eine sorgfältigere Handhabung von API-Änderungen, sowie der direkten und rückwärtskompatibilität;
• es müssen zusätzliche Runtime-Bibliotheken hinzugefügt werden, um den Betrieb der Geschäftslogik in jedem solchen Microservice zu gewährleisten, was neue Anforderungen an diese Bibliotheken mit sich bringt: Leichtgewichtigkeit und minimale transitive Abhängigkeiten;
• Entwickler von Geschäftslogik müssen die Versionen von Bibliotheken im Auge behalten: Wenn ein Mikrodienst längere Zeit nicht gewartet wurde, hat er wahrscheinlich veraltete Bibliotheken. Dies kann unerwartete Hindernisse beim Hinzufügen neuer Funktionen schaffen und könnte eine Migration der alten Geschäftslogik dieses Dienstes auf neue Versionen von Bibliotheken erforderlich machen, wenn zwischen den Versionen inkompatible Änderungen vorgenommen wurden.

Eine Schicht von Plattformdiensten ist in einer solchen Architektur ebenfalls vorhanden, jedoch bildet diese Schicht nicht den Container zur Ausführung der domänenspezifischen Geschäftslogik, sondern lediglich deren Umgebung, indem sie unterstützende „Plattform“-Funktionen bereitstellt. Eine solche Schicht ist notwendig, um nicht nur die Leichtgewichtigkeit der domänenbasierten Mikrodienste zu bewahren, sondern auch um die Verwaltung zu zentralisieren.

Zum Beispiel erzeugen benutzerdefinierte Aktivitäten in Geschäftsprozessen Aufgaben. Wenn der Benutzer jedoch mit Aufgaben arbeitet, sollte er alle Aufgaben aus verschiedenen Domänen in einer gemeinsamen Liste sehen können. Daher muss es einen entsprechenden Plattform-Service zur Registrierung von Aufgaben geben, der von der spezifischen Geschäftslogik der Domäne entkoppelt ist. Die Kapselung der Geschäftslogik in einem solchen Kontext zu bewahren, ist durchaus problematisch und stellt einen weiteren Kompromiss dieser Architektur dar.

Integration von Geschäftsprozessen aus der Sicht des Anwendungsentwicklers

Wie bereits erwähnt, muss der Anwendungsentwickler von den technischen und ingenieurtechnischen Merkmalen der Interaktion mehrerer Anwendungen abstrahiert werden, um eine hohe Produktivität der Entwicklung gewährleisten zu können.

Lassen Sie uns versuchen, eine recht anspruchsvolle Integrationsaufgabe zu lösen, die speziell für diesen Artikel entworfen wurde. Es wird eine "Spiel"-Aufgabe sein, die drei Anwendungen umfasst, von denen jede einen bestimmten Domänennamen definiert: "app1", "app2", "app3".

In jeder Anwendung laufen Geschäftsprozesse, die über eine Integrationsbus "Ball spielen". Die Nachrichten mit dem Namen "Ball" übernehmen die Rolle des Balls.

Spielregeln:

• Der erste Spieler ist der Initiator. Er lädt andere Spieler zum Spiel ein, beginnt das Spiel und kann es jederzeit beenden;
• Die anderen Spieler erklären ihre Teilnahme am Spiel, "lernen" einander und den ersten Spieler kennen;
• Haben sie den Ball angenommen, wählt der Spieler einen anderen teilnehmenden Spieler aus und übergibt ihm den Ball. Die Gesamtzahl der Pässe wird gezählt;
• Jeder Spieler hat "Energie", die mit jedem Pass dieses Spielers abnimmt. Wenn die Energie aufgebraucht ist, scheidet der Spieler aus dem Spiel aus und erklärt seinen Rückzug;
• Wenn ein Spieler alleine geblieben ist, erklärt er sofort seinen Rückzug;
• Wenn alle Spieler ausgeschieden sind, erklärt der erste Spieler das Spiel für beendet. Hat er zuvor das Spiel verlassen, bleibt er trotzdem als Beobachter, um das Spiel zu beenden.

Um diese Aufgabe zu lösen, werde ich unsere DSL für Geschäftsprozesse verwenden, die es ermöglicht, die Logik in Kotlin kompakt und mit minimalem Boilerplate zu beschreiben.

Im App1 wird der Geschäftsprozess des ersten Spielers (der gleichzeitig der Initiator des Spiels ist) laufen:

Klasse 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()

// Diese Klasse ist eine Prozessinstanz: kapselt ihren internen Zustand
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
}

// Dies ist die Prozessmodell-Deklaration: wird einmal erstellt, von allen
// Instanzen des entsprechenden Klassenprozesses verwendet
val initialPlayerModel = processModel(name = "InitialPlayer",
                                                     version = 1) {

    // Nach den Regeln ist der erste Spieler der Initiator des Spiels und muss der einzige sein
    uniqueConstraint = UniqueConstraints.singleton

    // Wir erklären die Aktivitäten, aus denen der Geschäftsprozess besteht
    val sendNewGameSignal = signal("NewGame")
    val sendStopGameSignal = signal("StopGame")
    val startTask = humanTask("Start") {
        taskOperation {
            processCondition { players.size > 0 }
            confirmation { "${players.size} Spieler haben sich verbunden. Beginnen wir?" }
        }
    }
    val stopTask = humanTask("Stop") {
        taskOperation {}
    }
    val waitPlayerJoin = signalWait("PlayerJoin") { signal ->
        players.add(PlayerInfo(
                signal.data!!,
                signal.sender.domain,
                signal.sender.processInstanceId))
        println("... Spieler ${signal.data} beigetreten ...")
    }
    val waitPlayerOut = signalWait("PlayerOut") { signal ->
        players.remove(PlayerInfo(
                signal.data!!,
                signal.sender.domain,
                signal.sender.processInstanceId))
        println("... Spieler ${signal.data} ist raus ...")
    }
    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
    }

    // Jetzt konstruieren wir den Prozessgraphen aus den erklärten Aktivitäten
    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)
    }

    // Wir hängen zusätzliche Handler für das Logging an die Aktivitäten
    sendNewGameSignal.onExit { println("Lasst uns spielen!") }
    sendStopGameSignal.onExit { println("Stop!") }
    sendPlayerOut.onExit { println("$playerName: Ich bin raus!") }
}

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

Neben der Ausführung der Geschäftslogik kann der angegebene Code ein objekthaftes Modell des Geschäftsprozesses liefern, das in Form eines Diagramms visualisiert werden kann. Für den Visualisierer haben wir bisher keine Umsetzung realisiert, daher mussten wir etwas Zeit mit dem Zeichnen verbringen (hier habe ich die BPMN-Notation in Bezug auf die Verwendung von Gates leicht vereinfacht, um die Konsistenz des Diagramms mit dem angegebenen Code zu verbessern):

Die Anwendung app2 wird den Geschäftsprozess eines anderen Spielers umfassen:

class RandomPlayer

import ru.krista.bpm.ProzessInstanz
import ru.krista.bpm.runtime.ProzessImpl
import ru.krista.bpm.runtime.dsl.prozessModell
import ru.krista.bpm.runtime.instance.NachrichtSendenInstanz

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

class SpielerListe: ArrayList<SpielerInfo>()

class ZufallsSpieler : ProzessImpl<ZufallsSpieler>(zufallsSpielerModell) {

    var spielerName: String by input(persistent = true, 
                                    defaultValue = "ZufallsSpieler")
    var energie: Int by input(persistent = true, defaultValue = 30)
    var spieler: SpielerListe by persistent(SpielerListe())
    var alleSpielerDraußen: Boolean by persistent(false)
    var schussZähler: Int = 0

    val eigenerSpieler: SpielerInfo
        get() = SpielerInfo(spielerName, env.eventDispatcher.domainName, id)
}

val zufallsSpielerModell = prozessModell<ZufallsSpieler>(name = "ZufallsSpieler", 
                                                   version = 1) {

    val warteAufNeuesSpielSignal = signalWarten<String>("NeuesSpiel")
    val warteAufStopSpielSignal = signalWarten<String>("StopSpiel")
    val sendeSpielerBeitritt = signal<String>("SpielerBeitritt") {
        signalDaten = { spielerName }
    }
    val sendeSpielerDraußen = signal<String>("SpielerDraußen") {
        signalDaten = { spielerName }
    }
    val warteAufSpielerBeitritt = signalWartenBenutzerdefiniert<String>("SpielerBeitritt") {
        ereignisBedingung = { signal ->
            signal.sender.prozessInstanzId != prozess.id 
                && !prozess.spieler.any { signal.sender.prozessInstanzId == it.id}
        }
        handler = { signal ->
            spieler.add(SpielerInfo(
                    signal.daten!!,
                    signal.sender.domain,
                    signal.sender.prozessInstanzId))
        }
    }
    val warteAufSpielerDraußen = signalWarten<String>("SpielerDraußen") { signal ->
        spieler.remove(SpielerInfo(
                signal.daten!!,
                signal.sender.domain,
                signal.sender.prozessInstanzId))
        alleSpielerDraußen = spieler.isEmpty()
    }
    val sendeHandshake = nachrichtSenden<String>("Handshake") {
        nachrichtDaten = { spielerName }
        aktivierung = {
            empfängerDomain = prozess.spieler.last().domain
            empfängerProzessInstanzId = prozess.spieler.last().id
        }
    }
    val empfangeHandshake = nachrichtWarten<String>("Handshake") { nachricht ->
        if (!spieler.any { nachricht.sender.prozessInstanzId == it.id}) {
            spieler.add(SpielerInfo(
                    nachricht.daten!!, 
                    nachricht.sender.domain, 
                    nachricht.sender.prozessInstanzId))
        }
    }
    val werfeBall = nachrichtSenden<Int>("Ball") {
        nachrichtDaten = { schussZähler + 1 }
        aktivierung = { wähleNächstenSpieler() }
        beimEintritt { energie -= 1 }
    }
    val warteAufBall = nachrichtWartenDaten<Int>("Ball") {
        schussZähler = it
    }

    starteVon(warteAufNeuesSpielSignal)
            .gabelung("hauptGabelung") {
                nächste(sendeSpielerBeitritt)
                        .zweig("hauptSchleife") {
                            fallsWahr { energie < 5 || alleSpielerDraußen }
                                    .nächste(sendeSpielerDraußen)
                                    .nächste(warteAufBall)
                            fallsSonst()
                                    .nächste(warteAufBall)
                                    .nächste(werfeBall)
                                    .schleife()
                        }
                nächste(warteAufSpielerBeitritt).nächste(sendeHandshake).nächste(warteAufSpielerBeitritt)
                nächste(warteAufSpielerDraußen).nächste(warteAufSpielerDraußen)
                nächste(empfangeHandshake).nächste(empfangeHandshake)
                nächste(warteAufStopSpielSignal).beenden()
            }

    sendeSpielerBeitritt.beimVerlassen { println("$spielerName: Ich bin hier!") }
    sendeSpielerDraußen.beimVerlassen { println("$spielerName: Ich bin draußen!") }
}

private fun NachrichtSendenInstanz<ZufallsSpieler, Int>.wähleNächstenSpieler() {
    val spieler = if (prozess.spieler.isNotEmpty()) 
        prozess.spieler.random() 
    else 
        prozess.eigenerSpieler
    empfängerDomain = spieler.domain
    empfängerProzessInstanzId = spieler.id
    println("Schritt ${prozess.schussZähler + 1}: " +
            "${prozess.spielerName} >>> ${spieler.name}")
}

Diagramm:

In der App app3 wird der Spieler ein etwas anderes Verhalten aufweisen: Statt zufällig den nächsten Spieler auszuwählen, agiert er nach dem Round-Robin-Algorithmus:

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}")
}

Ansonsten unterscheidet sich das Verhalten des Spielers nicht vom vorherigen, daher ändert sich das Diagramm nicht.

Jetzt ist ein Test erforderlich, um das alles auszuführen. Ich werde nur den Code des Tests angeben, um den Artikel nicht mit Boilerplate zu überladen (tatsächlich habe ich die zuvor für die Integrationstestung anderer Geschäftsprozesse erstellte Testumgebung verwendet):

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");
    // Jetzt müssen wir ein wenig warten, bis die Spieler sich "kennenlernen".
    // Schlafen durch sleep - eine schlechte Lösung, aber die einfachste. 
    // Machen Sie das nicht in ernsthaften Tests!
    Thread.sleep(1000);
    // Spiel starten und Benutzereingaben schließen
    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;
}

Wir starten den Test und schauen uns das Protokoll an:

Konsole Ausgabe

Der Schlüssel lock://app1/process/InitialPlayer wurde gesperrt.
Lass uns spielen!
Der Schlüssel lock://app1/process/InitialPlayer wurde freigegeben.
Spieler2: Ich bin hier!
Spieler3: Ich bin hier!
Spieler4: Ich bin hier!
Spieler5: Ich bin hier!
... Spieler Player2 beitreten ...
... Spieler Player4 beitreten ...
... Spieler Player3 beitreten ...
... Spieler Player5 beitreten ...
Schritt 1: Spieler1 >>> Spieler3
Schritt 2: Spieler3 >>> Spieler5
Schritt 3: Spieler5 >>> Spieler3
Schritt 4: Spieler3 >>> Spieler4
Schritt 5: Spieler4 >>> Spieler3
Schritt 6: Spieler3 >>> Spieler4
Schritt 7: Spieler4 >>> Spieler5
Schritt 8: Spieler5 >>> Spieler2
Schritt 9: Spieler2 >>> Spieler5
Schritt 10: Spieler5 >>> Spieler4
Schritt 11: Spieler4 >>> Spieler2
Schritt 12: Spieler2 >>> Spieler4
Schritt 13: Spieler4 >>> Spieler1
Schritt 14: Spieler1 >>> Spieler4
Schritt 15: Spieler4 >>> Spieler3
Schritt 16: Spieler3 >>> Spieler1
Schritt 17: Spieler1 >>> Spieler2
Schritt 18: Spieler2 >>> Spieler3
Schritt 19: Spieler3 >>> Spieler1
Schritt 20: Spieler1 >>> Spieler5
Schritt 21: Spieler5 >>> Spieler1
Schritt 22: Spieler1 >>> Spieler2
Schritt 23: Spieler2 >>> Spieler4
Schritt 24: Spieler4 >>> Spieler5
Schritt 25: Spieler5 >>> Spieler3
Schritt 26: Spieler3 >>> Spieler4
Schritt 27: Spieler4 >>> Spieler2
Schritt 28: Spieler2 >>> Spieler5
Schritt 29: Spieler5 >>> Spieler2
Schritt 30: Spieler2 >>> Spieler1
Schritt 31: Spieler1 >>> Spieler3
Schritt 32: Spieler3 >>> Spieler4
Schritt 33: Spieler4 >>> Spieler1
Schritt 34: Spieler1 >>> Spieler3
Schritt 35: Spieler3 >>> Spieler4
Schritt 36: Spieler4 >>> Spieler3
Schritt 37: Spieler3 >>> Spieler2
Schritt 38: Spieler2 >>> Spieler5
Schritt 39: Spieler5 >>> Spieler4
Schritt 40: Spieler4 >>> Spieler5
Schritt 41: Spieler5 >>> Spieler1
Schritt 42: Spieler1 >>> Spieler5
Schritt 43: Spieler5 >>> Spieler3
Schritt 44: Spieler3 >>> Spieler5
Schritt 45: Spieler5 >>> Spieler2
Schritt 46: Spieler2 >>> Spieler3
Schritt 47: Spieler3 >>> Spieler2
Schritt 48: Spieler2 >>> Spieler5
Schritt 49: Spieler5 >>> Spieler4
Schritt 50: Spieler4 >>> Spieler2
Schritt 51: Spieler2 >>> Spieler5
Schritt 52: Spieler5 >>> Spieler1
Schritt 53: Spieler1 >>> Spieler5
Schritt 54: Spieler5 >>> Spieler3
Schritt 55: Spieler3 >>> Spieler5
Schritt 56: Spieler5 >>> Spieler2
Schritt 57: Spieler2 >>> Spieler1
Schritt 58: Spieler1 >>> Spieler4
Schritt 59: Spieler4 >>> Spieler1
Schritt 60: Spieler1 >>> Spieler4
Schritt 61: Spieler4 >>> Spieler3
Schritt 62: Spieler3 >>> Spieler2
Schritt 63: Spieler2 >>> Spieler5
Schritt 64: Spieler5 >>> Spieler4
Schritt 65: Spieler4 >>> Spieler5
Schritt 66: Spieler5 >>> Spieler1
Schritt 67: Spieler1 >>> Spieler5
Schritt 68: Spieler5 >>> Spieler3
Schritt 69: Spieler3 >>> Spieler4
Schritt 70: Spieler4 >>> Spieler2
Schritt 71: Spieler2 >>> Spieler5
Schritt 72: Spieler5 >>> Spieler2
Schritt 73: Spieler2 >>> Spieler1
Schritt 74: Spieler1 >>> Spieler4
Schritt 75: Spieler4 >>> Spieler1
Schritt 76: Spieler1 >>> Spieler2
Schritt 77: Spieler2 >>> Spieler5
Schritt 78: Spieler5 >>> Spieler4
Schritt 79: Spieler4 >>> Spieler3
Schritt 80: Spieler3 >>> Spieler1
Schritt 81: Spieler1 >>> Spieler5
Schritt 82: Spieler5 >>> Spieler1
Schritt 83: Spieler1 >>> Spieler4
Schritt 84: Spieler4 >>> Spieler5
Schritt 85: Spieler5 >>> Spieler3
Schritt 86: Spieler3 >>> Spieler5
Schritt 87: Spieler5 >>> Spieler2
Schritt 88: Spieler2 >>> Spieler3
Spieler2: Ich bin raus!
Schritt 89: Spieler3 >>> Spieler4
... Spieler Player2 ist raus ...
Schritt 90: Spieler4 >>> Spieler1
Schritt 91: Spieler1 >>> Spieler3
Schritt 92: Spieler3 >>> Spieler1
Schritt 93: Spieler1 >>> Spieler4
Schritt 94: Spieler4 >>> Spieler3
Schritt 95: Spieler3 >>> Spieler5
Schritt 96: Spieler5 >>> Spieler1
Schritt 97: Spieler1 >>> Spieler5
Schritt 98: Spieler5 >>> Spieler3
Schritt 99: Spieler3 >>> Spieler5
Schritt 100: Spieler5 >>> Spieler4
Schritt 101: Spieler4 >>> Spieler5
Spieler4: Ich bin raus!
... Spieler Player4 ist raus ...
Schritt 102: Spieler5 >>> Spieler1
Schritt 103: Spieler1 >>> Spieler3
Schritt 104: Spieler3 >>> Spieler1
Schritt 105: Spieler1 >>> Spieler3
Schritt 106: Spieler3 >>> Spieler5
Schritt 107: Spieler5 >>> Spieler3
Schritt 108: Spieler3 >>> Spieler1
Schritt 109: Spieler1 >>> Spieler3
Schritt 110: Spieler3 >>> Spieler5
Schritt 111: Spieler5 >>> Spieler1
Schritt 112: Spieler1 >>> Spieler3
Schritt 113: Spieler3 >>> Spieler5
Schritt 114: Spieler5 >>> Spieler3
Schritt 115: Spieler3 >>> Spieler1
Schritt 116: Spieler1 >>> Spieler3
Schritt 117: Spieler3 >>> Spieler5
Schritt 118: Spieler5 >>> Spieler1
Schritt 119: Spieler1 >>> Spieler3
Schritt 120: Spieler3 >>> Spieler5
Schritt 121: Spieler5 >>> Spieler3
Spieler5: Ich bin raus!
... Spieler Player5 ist raus ...
Schritt 122: Spieler3 >>> Spieler5
Schritt 123: Spieler5 >>> Spieler1
Spieler5: Ich bin raus!
Schritt 124: Spieler1 >>> Spieler3
... Spieler Player5 ist raus ...
Schritt 125: Spieler3 >>> Spieler1
Schritt 126: Spieler1 >>> Spieler3
Spieler1: Ich bin raus!
... Spieler Player1 ist raus ...
Schritt 127: Spieler3 >>> Spieler3
Spieler3: Ich bin raus!
Schritt 128: Spieler3 >>> Spieler3
... Spieler Player3 ist raus ...
Spieler3: Ich bin raus!
Stop!
Schritt 129: Spieler3 >>> Spieler3
Spieler3: Ich bin raus!

Aus alledem lassen sich mehrere wichtige Schlussfolgerungen ziehen:

• Mit den richtigen Werkzeugen können Anwendungsentwickler Integrationsinteraktionen zwischen Anwendungen erstellen, ohne von der Geschäftslogik abzuweichen;
• Die Komplexität der Integrationsaufgabe, die Ingenieurkompetenzen erfordert, kann innerhalb des Frameworks verborgen werden, wenn dies von Anfang an in die Architektur des Frameworks integriert wird. Die Schwierigkeit der Aufgabe lässt sich jedoch nicht verbergen, weshalb die Lösung einer schwierigen Aufgabe im Code entsprechend aussehen wird;
• Bei der Entwicklung der Integrationslogik muss unbedingt die eventual consistency sowie das Fehlen der Linearität in der Zustandsänderung aller Integrationsbeteiligten berücksichtigt werden. Dies zwingt dazu, die Logik zu verkomplizieren, um sie unempfindlich gegenüber der Reihenfolge der eintreffenden externen Ereignisse zu gestalten. In unserem Beispiel muss der Spieler erst an dem Spiel teilnehmen, nachdem er seinen Rücktritt erklärt hat: andere Spieler werden weiterhin den Ball zu ihm passen, bis die Information über seinen Rücktritt alle Beteiligten erreicht und verarbeitet ist. Diese Logik ergibt sich nicht aus den Spielregeln und ist ein Kompromiss innerhalb der gewählten Architektur.

Als Nächstes werden wir verschiedene Feinheiten unserer Lösung, Kompromisse und andere Aspekte besprechen.

Alle Nachrichten – in einer Warteschlange

Alle integrierbaren Anwendungen arbeiten mit einer integrierten Schnittstelle, die als externer Broker, einer BPMQueue für Nachrichten und einem BPMTopic für Signale (Ereignisse) dargestellt wird. Das Durchleiten aller Nachrichten über eine einzige Warteschlange ist in der Tat ein Kompromiss. Auf der Ebene der Geschäftslogik können nun beliebig viele neue Nachrichtentypen eingeführt werden, ohne die Struktur des Systems zu ändern. Das ist eine erhebliche Vereinfachung, bringt jedoch auch gewisse Risiken mit sich, die uns im Kontext unserer typischen Aufgaben nicht allzu bedeutend erschienen.

Es gibt jedoch einen wichtigen Punkt: Jede Anwendung filtert ihre "eigenen" Nachrichten bereits beim Eingang nach ihrem Domänennamen. Der Domänenname kann auch in Signalen angegeben werden, wenn die "Sichtbarkeit" des Signals auf eine einzige Anwendung beschränkt werden soll. Dies sollte die Durchsatzkapazität der Schnittstelle erhöhen, aber die Geschäftslogik muss jetzt mit Domänennamen umgehen: Für die Adressierung von Nachrichten ist dies erforderlich, für Signale wünschenswert.

Gewährleistung der Zuverlässigkeit der Schnittstelle

Die Zuverlässigkeit setzt sich aus mehreren Aspekten zusammen:

• Der ausgewählte Nachrichtenbroker ist ein kritischer Bestandteil der Architektur und ein Einzelpunkt des Fehlers: Er muss ausreichend ausfallsicher sein. Es sollten nur bewährte Implementierungen mit solidem Support und einer großen Community verwendet werden;
• Es muss eine hohe Verfügbarkeit des Nachrichtenbrokers gewährleistet sein, wofür er physisch von den integrierten Anwendungen getrennt sein muss (eine hohe Verfügbarkeit von Anwendungen mit geschäftlicher Logik ist wesentlich komplizierter und teurer sicherzustellen);
• Der Broker muss Garantien für die Zustellung "at least once" bieten. Dies ist eine zwingende Anforderung für einen zuverlässigen Betrieb des Integrationsbusses. Bei Garantien der Ebenen "exactly once" besteht keine Notwendigkeit: Geschäftsprozesse sind in der Regel nicht empfindlich gegenüber der Mehrfachanlieferung von Nachrichten oder Ereignissen, und bei speziellen Aufgaben, in denen dies wichtig ist, ist es einfacher, eine zusätzliche Überprüfung in die Geschäftslogik einzufügen, als ständig ausreichend "teure" Garantien zu verwenden;
• Die Sendung von Nachrichten und Signalen sollte in die gesamte Transaktion zur Zustandsänderung der Geschäftsprozesse und der Domänendaten einbezogen werden. Bevorzugt wäre der Einsatz eines Musters Transaktionale Outbox, aber es erfordert eine zusätzliche Tabelle in der Datenbank und einen Repeater. In JEE-Anwendungen kann dieser Aspekt durch die Verwendung eines lokalen JTA-Managers vereinfacht werden, jedoch muss die Verbindung zu dem ausgewählten Broker im Modus arbeiten können XA;
• Eingehende Nachrichten und Ereignisse müssen ebenfalls mit der Transaktion zur Änderung des Geschäftsprozesszustands arbeiten: Wenn eine solche Transaktion zurückgesetzt wird, muss auch der Empfang der Nachricht storniert werden;
• Nachrichten, die aufgrund von Fehlern nicht zugestellt werden konnten, müssen in einem separaten Speicher abgelegt werden DLQ (Dead Letter Queue). Zu diesem Zweck haben wir einen separaten plattformbasierten Mikrodienst erstellt, der solche Nachrichten in seinem Speicher speichert, sie nach Attributen indiziert (für schnelle Gruppierung und Suche) und eine API für die Anzeige, das erneute Senden an die Zieladresse und das Löschen von Nachrichten bereitstellt. Systemadministratoren können über ihre Webschnittstelle mit diesem Dienst arbeiten;
• In den Broker-Einstellungen sollte die Anzahl der Wiederholungsversuche und die Verzögerungen zwischen den Lieferungen angepasst werden, um die Wahrscheinlichkeit zu verringern, dass Nachrichten in die DLQ gelangen (die optimalen Parameter zu berechnen, ist praktisch unmöglich, aber man kann empirisch handeln und sie während des Betriebs anpassen);
• Der DLQ-Speicher sollte kontinuierlich überwacht werden, und das Überwachungssystem sollte die Systemadministratoren benachrichtigen, damit sie so schnell wie möglich auf unzustellbare Nachrichten reagieren können. Dies wird helfen, die "Zonen des Einflusses" eines aufgetretenen Fehlers oder einer Geschäftslogik-Fehlfunktion zu verringern;
• Die Integrationsbus sollte unempfindlich gegenüber vorübergehenden Ausfällen von Anwendungen sein: die Abonnements für das Thema sollten dauerhaft sein, und der Anwendungsname sollte einzigartig sein, damit während der Abwesenheit der Anwendung ihre Nachrichten aus der Warteschlange nicht von jemand anderem verarbeitet werden.

Gewährleistung der Thread-Sicherheit der Geschäftslogik

Einzelne Instanzen von Geschäftsprozessen können mehrere Nachrichten und Ereignisse gleichzeitig empfangen, deren Verarbeitung parallel gestartet wird. Für den Anwendungsentwickler muss alles jedoch einfach und thread-sicher sein.

Die Geschäftslogik des Prozesses verarbeitet jedes externe Ereignis, das diesen Geschäftsprozess beeinflusst, einzeln. Solche Ereignisse können sein:

• der Start einer Instanz des Geschäftsprozesses;
• eine Benutzeraktion, die sich auf eine Aktivität innerhalb des Geschäftsprozesses bezieht;
• der Eingang einer Nachricht oder eines Signals, auf das die Instanz des Geschäftsprozesses aboniert ist;
• das Auslösen eines von der Instanz des Geschäftsprozesses festgelegten Timers;
• steuernde Eingriffe über die API (z. B. ein Notstop des Prozesses).

Jedes Ereignis kann den Zustand einer Instanz des Geschäftsprozesses ändern: Einige Aktivitäten können beendet und andere gestartet werden, und die Werte persistenter Eigenschaften können sich ändern. Der Abschluss einer Aktivität kann die Aktivierung einer oder mehrerer nachfolgender Aktivitäten zur Folge haben. Diese können wiederum auf das Eintreffen anderer Ereignisse warten oder, falls keine zusätzlichen Daten benötigt werden, im gleichen Transaktionsrahmen abgeschlossen werden. Vor dem Abschluss der Transaktion wird der neue Zustand des Geschäftsprozesses in der Datenbank gespeichert, wo er auf das Eintreffen des nächsten externen Ereignisses warten wird.

Persistente Daten des Geschäftsprozesses, die in eine relationale Datenbank gespeichert sind, bieten einen praktischen Synchronisationspunkt für die Verarbeitung, wenn SELECT FOR UPDATE verwendet wird. Wenn es einer Transaktion gelingt, den Zustand des Geschäftsprozesses aus der Datenbank abzurufen, um ihn zu ändern, kann keine andere Transaktion parallel denselben Zustand für eine andere Änderung abrufen. Nach Abschluss der ersten Transaktion erhält die zweite Transaktion garantiert den bereits geänderten Zustand.

Durch den Einsatz von pessimistischen Sperren auf Datenbankseite erfüllen wir alle notwendigen Anforderungen ACID, während wir gleichzeitig die Möglichkeit zur Skalierung der Anwendung mit Geschäftslogik durch Erhöhung der Anzahl der gestarteten Instanzen beibehalten.

Allerdings birgt die Verwendung pessimistischer Sperren die Gefahr von Deadlocks. Daher sollte SELECT FOR UPDATE tatsächlich mit einem vernünftigen Timeout begrenzt werden, um im Falle von Deadlocks in offensichtlichen Fällen in der Geschäftslogik gewappnet zu sein.

Ein weiteres Problem ist die Synchronisation des Starts von Geschäftsprozessen. Solange keine Instanz des Geschäftsprozesses existiert, gibt es auch keinen Status in der Datenbank, weshalb die beschriebene Methode nicht geeignet ist. Wenn die Einzigartigkeit der Instanz des Geschäftsprozesses in einem bestimmten Scope gewährleistet werden muss, ist ein Synchronisationsobjekt erforderlich, das mit der Prozessklasse und dem entsprechenden Scope assoziiert ist. Um dieses Problem zu lösen, verwenden wir einen anderen Sperrmechanismus, der es ermöglicht, eine Sperre auf eine beliebige Ressource zu setzen, die über einen URI-Schlüssel definiert ist, über einen externen Dienst.

In unseren Beispielen enthält der Geschäftsprozess InitialPlayer die Deklaration

uniqueConstraint = UniqueConstraints.singleton

Daher erscheinen im Protokoll Nachrichten über das Sperren und Freigeben des entsprechenden Schlüssels. Bei anderen Geschäftsprozessen gibt es solche Nachrichten nicht: uniqueConstraint ist nicht festgelegt.

Probleme von Geschäftsprozessen mit persistentem Zustand

Manchmal hilft ein persistenter Zustand nicht nur, sondern hindert auch die Entwicklung erheblich.
Probleme treten auf, wenn Änderungen an der Geschäftslogik und/oder dem Modell des Geschäftsprozesses vorgenommen werden müssen. Nicht jede solche Änderung ist mit dem alten Zustand der Geschäftsprozesse kompatibel. Wenn in der Datenbank viele „lebende“ Instanzen vorhanden sind, kann das Einbringen inkompatibler Änderungen viele Probleme verursachen, mit denen wir oft bei der Verwendung von jBPM konfrontiert waren.

Je nach Tiefe der Änderungen gibt es zwei Möglichkeiten:

1. einen neuen Typ von Geschäftsprozess zu erstellen, um keine inkompatiblen Änderungen am alten vorzunehmen und ihn bei der Ausführung neuer Instanzen anstelle des alten zu verwenden. Die alten Instanzen werden weiterhin „wie gewohnt“ arbeiten;
2. den persistenten Zustand der Geschäftsprozesse bei der Aktualisierung der Geschäftslogik zu migrieren.

Der erste Ansatz ist einfacher, hat jedoch seine Einschränkungen und Nachteile, zum Beispiel:

• die doppelte Geschäftslogik in vielen Geschäftsprozessmodellen, was zu einem Anstieg des Umfangs der Geschäftslogik führt;
• häufig ist ein sofortiger Übergang zur neuen Geschäftslogik erforderlich (insbesondere bei Integrationsaufgaben – nahezu immer);
• der Entwickler weiß nicht, wann veraltete Modelle gelöscht werden können.

In der Praxis verwenden wir beide Ansätze, haben jedoch eine Reihe von Entscheidungen getroffen, um uns das Leben zu erleichtern:

• In der Datenbank wird der persistente Zustand des Geschäftsprozesses in einem leicht lesbaren und leicht verarbeitbaren Format gespeichert: in einer JSON-Zeile. Dies ermöglicht Migrationen sowohl innerhalb der Anwendung als auch extern. Im äußersten Fall kann man auch manuell nachhelfen (besonders nützlich während der Entwicklung und beim Debugging);
• Die Integrationsgeschäftslogik verwendet keine Namen von Geschäftsprozessen, sodass jederzeit die Implementierung eines der beteiligten Prozesse durch eine neue mit einem neuen Namen (z. B. "InitialPlayerV2") ersetzt werden kann. Die Bindung erfolgt über die Namen von Nachrichten und Signalen;
• Das Prozessmodell hat eine Versionsnummer, die wir erhöhen, wenn wir inkompatible Änderungen an diesem Modell vornehmen. Diese Nummer wird zusammen mit dem Status der Prozesseinstanz gespeichert.
• Der persistente Status des Prozesses wird zunächst in ein praktisches Objektmodell aus der Datenbank eingelesen, mit dem die Migrationsprozedur arbeiten kann, wenn sich die Versionsnummer des Modells geändert hat.
• Die Migrationsprozedur wird in der Nähe der Geschäftslogik platziert und „faul“ für jede Instanz des Geschäftsprozesses zum Zeitpunkt ihrer Wiederherstellung aus der Datenbank aufgerufen.
• Wenn der Status aller Prozesseinstanzen schnell und synchron migriert werden muss, kommen klassischere Datenbankmigrationstechniken zum Einsatz, wobei jedoch mit JSON gearbeitet werden muss.

Wird ein weiteres Framework für Geschäftsprozesse benötigt?

Die in dem Artikel beschriebenen Lösungen haben es uns ermöglicht, unser Leben erheblich zu erleichtern, den Kreis der Fragen, die auf der Ebene der Anwendungsentwicklung gelöst werden, zu erweitern und Ideen zur Trennung der Geschäftslogik in Microservices attraktiver zu gestalten. Dafür wurde viel Arbeit geleistet, ein sehr ‚leichtgewichtiger‘ Framework für Geschäftsprozesse erstellt sowie Dienstkomponenten zur Lösung der genannten Probleme im Kontext einer breiten Palette von Anwendungsaufgaben entwickelt. Wir möchten diese Ergebnisse gerne teilen und die Entwicklung gemeinsamer Komponenten unter einer freien Lizenz veröffentlichen. Dies wird einige Anstrengungen und Zeit erfordern. Ein Verständnis für die Nachfrage nach solchen Lösungen könnte für uns einen zusätzlichen Anreiz darstellen. In dem vorgeschlagenen Artikel wird den Möglichkeiten des Frameworks nur wenig Aufmerksamkeit geschenkt, aber einige von ihnen sind aus den vorgestellten Beispielen ersichtlich. Falls wir unser Framework dennoch veröffentlichen, wird es einen eigenen Artikel dazu geben. Bis dahin wären wir dankbar, wenn Sie uns ein kleines Feedback hinterlassen und die Frage beantworten:

Nur registrierte Benutzer können an der Umfrage teilnehmen. Bitte melden Sie sich an.Sind Sie an Contour interessiert?

Brauchen wir noch ein weiteres Framework für Geschäftsprozesse?

• 18,8%Ja, wir suchen schon lange nach etwas Ähnlichem.

• 12,5%ich würde gerne mehr über Ihre Implementierung erfahren, das könnte nützlich sein

• 6,2%wir verwenden eines der bestehenden Frameworks, denken aber über einen Wechsel nach

• 18,8%wir verwenden eines der bestehenden Frameworks, sind damit zufrieden

• 18,8%wir kommen ohne Framework aus

• 25,0%wir entwickeln unser eigenes

16 Nutzer haben abgestimmt. 7 Nutzer haben sich enthalten.

Quelle: habr.com