[JAVA] Versuchen Sie es mit dem Axon Framework

Ich hatte zufällig die Möglichkeit, einen internen Geschäftsablauf zu implementieren. Es ist wie Antrag → Genehmigung durch die betroffene Person → Genehmigung durch den endgültigen Manager.

Normalerweise können Sie mit der CRUD-Architektur mit MVC of Spring entwickeln, aber ich möchte daraus eine CQRS + Event Sourcing-Architektur machen, an der ich mich schon lange interessiert habe, und ich werde das Axon Framework evaluieren, das dies realisiert.

Axon Framework

• CQRS-Framework.
• Bietet auch Event Sourcing-Funktion.
• Event Sourcing ist nicht erforderlich.
• Bietet auch eine Funktion zum Verknüpfen mit Spring Boot.
• Spring-Boot-Devtools sind jedoch nicht kompatibel und können nicht verwendet werden, und die Entwicklungseffizienz ist etwas schlechter.
• Die neueste Version zum Zeitpunkt des Schreibens dieses Artikels ist 4.3.
• Apache License v2.0

Darüber hinaus wurde ab Version 4 ein Server für die Microservices-Umgebung namens Axon Server hinzugefügt, und es wird anscheinend empfohlen, mit Axon Server und Axon Framework zusammenzuarbeiten. Diesmal werde ich jedoch nicht damit umgehen, da ich Axon Framework verstehen möchte.

die Architektur

Unten finden Sie ein Diagramm der Version 3-Dokumentation, das (glaube ich) leichter zu verstehen ist als das neueste Diagramm.

• Entwickler implementieren Command Handler, Event Handler und Thin Data Layer. Die Benutzeroberfläche hat nichts mit Axon zu tun, daher werde ich sie weglassen.
• Wenn der Befehlshandler ein Ereignis gemäß der Geschäftslogik sendet, zeichnet Axon das Ereignis im Ereignisspeicher auf und ruft den dem Ereignis entsprechenden Ereignishandler auf.
• Der Ereignishandler am unteren Rand der Abbildung generiert Daten für die Abfrage. Welche Art von Datenspeicher verwendet werden soll, ist optional.

Verarbeitungsreihenfolge

Dies ist ein Bild, das aus Axons Dokumentation und Beispielen hervorgeht.

Befehl

1. Senden Sie Befehle über den von Axon bereitgestellten Befehlsbus an Aggregate.
2. Eine dem Befehl entsprechende Instanz von Aggregate wird erstellt. Wenn das entsprechende Aggregate-Ereignis bereits aufgezeichnet wurde, wird das Ereignis auf die Instanz angewendet (abgespielt).
3. Aggregate empfängt den Befehl und sendet ein Ereignis entsprechend dem Status von Aggregate an Event Bus.
4. Der Ereignisbus zeichnet das Ereignis auf und sendet es an Aggregate.
5. Aggregat ändert seinen Status je nach Ereignis in den Inhalt.

Abfrage

1. Der Ereignisprozessor erkennt, dass das Ereignis im Ereignisspeicher aufgezeichnet wurde, und sendet das Ereignis an die dünne Datenschicht.
2. Thin Data Layer speichert Daten in einer Form, die leicht abzufragen ist.
3. Abfragen werden über den Abfragebus an die dünne Datenschicht gesendet.
4. Thin Data Layer ruft Daten aus dem Datenspeicher ab und gibt sie zurück.

Stichprobe

Lassen Sie uns zunächst einen Prozess implementieren, der nur die Anwendung zur Evaluierung CRUDs.

1. Abhängigkeiten

Fügen Sie der Spring Boot-Anwendung den Axon-Spring-Boot-Starter (1) hinzu. Da wir diesmal Axon Server nicht verwenden, schließen wir Axon-Server-Connector aus. Andernfalls erhalten Sie einen Fehler oder eine Warnung bezüglich eines Verbindungsfehlers zu Axon Server, während die Anwendung ausgeführt wird.

JPA und JDBC werden als Engines für den Zugriff auf den Event Store im Axon Framework bereitgestellt. Da die Verarbeitung zum Ereignisspeicher vom Axon Framework ausgeblendet wird, ist die Auswahl ein Kompromiss mit der Implementierungsmethode von Thin Data Layer. Dieses Mal haben wir der JPA-Engine Spring-Boot-Starter-Data-JPA (②) hinzugefügt, was nur minimale Einstellungen erfordert.

• In der tatsächlichen Entwicklung möchte ich Jooq verwenden, was mir persönlich gefällt. In diesem Fall verwende ich die JDBC-Engine.

pom.xml(Auszug)

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.6.RELEASE</version>
    <relativePath/>
  </parent>

  <properties>
    <java.version>11</java.version>
    <kotlin.version>1.3.71</kotlin.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.axonframework</groupId>
      <artifactId>axon-spring-boot-starter</artifactId> <!-- ① -->
      <version>4.3.2</version>
      <exclusions>
        <exclusion>
          <groupId>org.axonframework</groupId>
          <artifactId>axon-server-connector</artifactId>
        </exclusion>
      </exclusions>
    </dependency>

    <!-- Web -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-thymeleaf</artifactId>
    </dependency>

    <!-- DB -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-data-jpa</artifactId> <!-- ② -->
    </dependency>
    <dependency>
      <groupId>com.h2database</groupId>
      <artifactId>h2</artifactId>
      <scope>runtime</scope>
    </dependency>

２. Aggregate Dies ist der zentrale Teil der Geschäftslogik im Axon Framework. Da Aggregat in DDD als Transaktionsgrenze bezeichnet wird, ist es wahrscheinlich, dass Ereignisse auf Aggregatbasis verwaltet werden.

2.1 Befehle und Ereignisse

Geben Sie im Befehl die Kennung des Aggregats an, das in der Eigenschaft gesendet werden soll, der @TargetAggregateIdentifier (1) zugewiesen wurde. Andererseits gibt es keine solche Spezifikation, da das Ereignis an dieselbe Aggregatinstanz gesendet wird, die den Befehl empfangen hat, oder an die instanzunabhängige Thin Data Layer.

message.kt

interface IInvoiceBody {
  val invoiceTitle: String
}

interface IInvoiceHeader {
  val invoiceId: UUID
}

interface IInvoice: IInvoiceHeader, IInvoiceBody

data class CreateInvoiceCommand(
  @field:TargetAggregateIdentifier // ①
  override val invoiceId: UUID,
  val body: IInvoiceBody): IInvoiceBody by body, IInvoice

data class UpdateInvoiceCommand(
  @field:TargetAggregateIdentifier // ①
  override val invoiceId: UUID,
  val body: IInvoiceBody): IInvoiceBody by body, IInvoice

data class RemoveInvoiceCommand(
  @field:TargetAggregateIdentifier // ①
  override val invoiceId: UUID
): IInvoiceHeader

data class InvoiceCreatedEvent(private val invoice: IInvoice): IInvoice by invoice

data class InvoiceUpdatedEvent(private val invoice: IInvoice): IInvoice by invoice

data class InvoiceRemovedEvent(
  override val invoiceId: UUID
): IInvoiceHeader

• Der Grund, warum wir häufig Schnittstellen und Delegaten verwenden, liegt darin, dass wir die Übertragung von Eigenschaftswerten von Befehlen zu Ereignissen unterbrechen möchten. Dies bedeutet nicht, dass dies nicht funktioniert.

２.2. Aggregate Weisen Sie der Eigenschaft, die @TargetAggregateIdentifier des Befehls entspricht, @AggregateIdentifier (1) zu. @Aggregate (②) ist eine Annotation für die Spring-Kooperation und hat auch die Funktion der @Component-Annotation.

Weisen Sie der Methode, die den Befehl verarbeitet, @CommandHandler (③) zu. Der Befehl zum Erstellen eines neuen Aggregats wird vom Konstruktor empfangen. Der Standardkonstruktor ist jedoch auch für die Verarbeitung nachfolgender Befehle erforderlich, da das Ereignis erst nach der Instanziierung wiedergegeben wird.

Das Ereignis wird von der Methode mit @EventSourcingHandler (④) empfangen und der Aggregatstatus wird aktualisiert. Rufen Sie AggregateLifecycle.markDeleted () am Ende des Aggregate-Lebensereignisses auf.

InvoiceAggregate.kt

@Aggregate // ②
class InvoiceAggregate(): IInvoice {
  @AggregateIdentifier // ①
  override lateinit var invoiceId: UUID
  override lateinit var invoiceTitle: String

  @CommandHandler // ③
  constructor(command: CreateInvoiceCommand): this() {
    AggregateLifecycle.apply(InvoiceCreatedEvent(command))
  }

  @EventSourcingHandler // ④
  fun on(event: InvoiceCreatedEvent) {
    invoiceId = event.invoiceId
    invoiceTitle = event.invoiceTitle
  }

  @CommandHandler // ③
  fun handle(command: UpdateInvoiceCommand) {
    AggregateLifecycle.apply(InvoiceUpdatedEvent(command))
  }

  @EventSourcingHandler // ④
  fun on(event: InvoiceUpdatedEvent) {
    invoiceTitle = event.invoiceTitle
  }

  @CommandHandler // ③
  fun handle(command: RemoveInvoiceCommand) {
    AggregateLifecycle.apply(InvoiceRemovedEvent(command.invoiceId))
  }

  @EventSourcingHandler // ④
  fun on(event: InvoiceRemovedEvent) {
    AggregateLifecycle.markDeleted() //⑤
  }
}

3. Thin Data Layer Im Ereignisspeicher aufgezeichnete Ereignisse werden mit @EventHandler (1) an die Methode übermittelt. Bitte beachten Sie, dass es sich von @EventSourcingHandler unterscheidet. Die Abfrage wird von der Methode mit @QueryHandler (②) empfangen, und die Daten, die die Bedingungen erfüllen, werden erfasst und zurückgegeben.

InvoiceService.kt

@Service
class InvoiceService(val invoiceRepo: InvoiceRepository) {
  class AllInvoicesQuery

  @QueryHandler // ②
  fun handle(query: AllInvoicesQuery): List<InvoiceEntity> {
    return invoiceRepo.findAll()
  }
  @EventHandler // ①
  fun on(event: InvoiceCreatedEvent) {
    invoiceRepo.save(InvoiceEntity(event.invoiceId, event.invoiceTitle))
  }
  @EventHandler // ①
  fun on(event: InvoiceUpdatedEvent) {
    invoiceRepo.save(InvoiceEntity(event.invoiceId, event.invoiceTitle))
  }
  @EventHandler // ①
  fun on(event: InvoiceRemovedEvent) {
    invoiceRepo.deleteById(event.invoiceId)
  }
}

4. Controller Dies ist der Controller in Spring MVC, der HTTP-Anforderungen in Befehle übersetzt und diese über Command Gateway und Query Gateway an Command Bus und Query Bus sendet. Das Senden von Befehlen an CommandGateway verfügt über eine asynchrone Sendemethode und eine synchrone sendAndWait-Methode. Hier verwende ich das PRG-Muster, aber da es zu Überholmanövern kommen kann, verwende ich sendAndWait.

InvoiceController.kt

@Controller
class InvoiceController(val commandGateway: CommandGateway, val queryGateway: QueryGateway) {

  companion object {
    const val REDIRECT_URL = "${UrlBasedViewResolver.REDIRECT_URL_PREFIX}/"
  }

  data class InvoiceBody(override val invoiceTitle: String): IInvoiceBody
  data class InvoiceRequest(
    override val invoiceId: UUID,
    override val invoiceTitle: String
  ): IInvoiceHeader, IInvoiceBody

  @GetMapping("/")
  fun topPage(model: Model): String {
    val invoices = queryGateway.query(InvoiceService.AllInvoicesQuery(), MultipleInstancesResponseType(InvoiceEntity::class.java)).get()
    model.addAttribute("invoices", invoices)
    return "index"
  }

  @PostMapping("/invoices")
  fun createInvoice(invoice: InvoiceBody): String {
    commandGateway.sendAndWait<Any>(CreateInvoiceCommand(UUID.randomUUID(), invoice))
    return REDIRECT_URL
  }

  @PostMapping("/invoices/update")
  fun updateInvoice(invoice: InvoiceRequest): String {
    commandGateway.sendAndWait<Any>(UpdateInvoiceCommand(invoice.invoiceId, invoice))
    return REDIRECT_URL
  }

  @PostMapping("/invoices/delete")
  fun deleteInvoice(@RequestParam invoiceId: UUID): String {
    commandGateway.sendAndWait<Any>(RemoveInvoiceCommand(invoiceId))
    return REDIRECT_URL
  }
}

Ärger

Ich kann das Ergebnis des Befehls in der Abfrage nicht sehen

Im Beispiel konnte ich im Beispiel die Daten nicht abrufen, selbst wenn ich eine Anwendung (/ Rechnungen) erstellt und zur Anzeige der Seite (/) umgeleitet habe. Beim Debuggen hat die @ CommandHandler-Methode ein Ereignis gesendet und die @ EventSourcingHandler-Methode wurde sofort aufgerufen, die @ EventHandler-Methode jedoch etwas später. Anscheinend läuft Axons Standard-Ereignisprozessor TrackingEventProcessor in einem separaten Thread und überwacht das Hinzufügen von Ereignissen mit dem Standardwert von 5 Sekunden.

Es ist wahrscheinlich eine gute Idee, den Überwachungszyklus für große Systeme zu optimieren, aber für kleinere ist es besser, zu einem anderen SubscribingEventProcessor zu wechseln.

@SpringBootApplication
class SampleAxonApplication {
  @Autowired
  fun configure(configurer: EventProcessingConfigurer) {
    configurer.usingSubscribingEventProcessors()
  }
}

Mit SubscribingEventProcessor wird die Methode von @ EventHandler vom selben Thread wie Controller aufgerufen, sodass Sie das Überholen von Abfragen vermeiden können.

Impressionen

Da verschiedene Untersuchungen noch nicht abgeschlossen sind, ist noch unklar, ob sie der tatsächlichen Entwicklung standhalten können. Zum Beispiel

• Wenn Sie die Implementierung von Aggregate aufgrund einer Spezifikationsänderung ändern, wurde das Ereignis bereits aufgezeichnet. Was soll ich also tun?
• Wenn ein Fehler ein falsches Ereignis aufzeichnet, kann das Ereignis geändert oder gelöscht werden?
• Ich möchte einen Operationsverlauf, aber gibt es einen Mechanismus, um eine Liste von Ereignissen abzurufen?
• Wenn Sie die Eindeutigkeit zwischen Aggregaten sicherstellen möchten, z. B. den Anwendungsnamen, wird dieser anscheinend im Datenspeicher überprüft, bevor der Befehl vom Controller gesendet wird. Ich bin jedoch ein wenig besorgt, ob dies allein möglich ist.

Und so weiter. Ich möchte diese in Zukunft untersuchen.