[JAVA] Essayez d'utiliser le Framework Axon

J'ai eu l'opportunité de mettre en place un business flow en interne. C'est comme une demande → approbation par la personne concernée → approbation par le responsable final.

Normalement, vous pouvez développer avec une architecture CRUD avec MVC de Spring, mais je veux en faire une architecture CQRS + Event Sourcing qui m'intéresse depuis longtemps, et j'évaluerai le Framework Axon qui le réalise.

Axon Framework

• Cadre CQRS.
• Fournit également la fonction Event Sourcing.
• La recherche d'événements n'est pas requise.
• Fournit également une fonction de liaison avec Spring Boot.
• Mais spring-boot-devtools est incompatible et ne peut pas être utilisé, et l'efficacité du développement est un peu inférieure.
• La dernière version au moment de la rédaction de cet article est la 4.3.
• Apache License v2.0

De plus, à partir de la version 4, un serveur pour l'environnement Microservices appelé Axon Server a été ajouté, et il semble qu'il soit recommandé de travailler avec Axon Server et Axon Framework, mais cette fois je ne le gérerai pas car je veux comprendre Axon Framework.

architecture

Vous trouverez ci-dessous un diagramme de la documentation de la version 3, qui est plus facile à comprendre (je pense) que le dernier diagramme.

• Les développeurs implémentent le gestionnaire de commandes, le gestionnaire d'événements et la couche de données mince. L'interface utilisateur n'a rien à voir avec Axon, je vais donc l'omettre.
• Lorsque le gestionnaire de commandes envoie un événement selon la logique métier, Axon enregistre l'événement dans le magasin d'événements et appelle le gestionnaire d'événements correspondant à l'événement.
• Le gestionnaire d'événements en bas de la figure génère des données pour la requête. Le type de magasin de données à utiliser est facultatif.

Séquence de traitement

Ceci est une image comprise à partir de la documentation et des exemples d'Axon.

commander

1. Envoyez des commandes à Aggregate via le bus de commande fourni par Axon.
2. Une instance d'Agrégat correspondant à la commande est créée. Si l'événement Aggregate correspondant a déjà été enregistré, l'événement sera appliqué (joué) à l'instance.
3. Aggregate reçoit la commande et envoie un événement au bus d'événements en fonction de l'état de l'agrégat.
4. Event Bus enregistre l'événement et l'envoie à Aggregate.
5. L'agrégat change son état en contenu en fonction de l'événement.

Requete

1. Le processeur d'événements détecte que l'événement a été enregistré dans le magasin d'événements et l'envoie à la couche de données mince.
2. Thin Data Layer stocke les données sous une forme facile à interroger.
3. Les requêtes sont envoyées au Thin Data Layer via Query Bus.
4. Thin Data Layer récupère les données du magasin de données et les renvoie.

échantillon

Tout d'abord, implémentons un processus qui ne fait que CRUDs l'application pour l'évaluation.

1. Dépendances

Ajoutez axon-spring-boot-starter (1) à l'application Spring Boot. Puisque nous n'utiliserons pas Axon Server cette fois, nous exclurons axon-server-connector. Sinon, vous obtiendrez une erreur ou un avertissement concernant l'échec de la connexion à Axon Server pendant que l'application est en cours d'exécution.

JPA et JDBC sont fournis en tant que moteurs pour accéder au magasin d'événements dans Axon Framework. Étant donné que le traitement vers le magasin d'événements est masqué par Axon Framework, celui à choisir sera un compromis avec la méthode de mise en œuvre de Thin Data Layer. Cette fois, nous avons ajouté spring-boot-starter-data-jpa (②) au moteur JPA, ce qui nécessite une configuration minimale.

• En développement réel, je souhaite utiliser Jooq, ce que j'aime personnellement, donc dans ce cas j'utiliserai le moteur JDBC.

pom.xml(Extrait)

  <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 C'est la partie centrale de la logique métier dans Axon Framework. Puisque l'agrégat est considéré comme une limite de transaction dans DDD, il est probable que les événements seront gérés sur une base agrégée.

2.1. Commandes et événements

Dans la commande, spécifiez l'identifiant de l'agrégat à envoyer dans la propriété à laquelle @TargetAggregateIdentifier (1) a été affecté. D'autre part, il n'y a pas de telle spécification car l'événement est envoyé à la même instance d'Agrégat qui a reçu la commande, ou à la couche de données légère indépendante de l'instance.

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

• La raison pour laquelle nous utilisons beaucoup les interfaces et les délégués est que nous voulons interrompre le transfert des valeurs de propriété des commandes aux événements, et cela ne signifie pas que cela ne fonctionnera pas.

２.2. Aggregate Attribuez @AggregateIdentifier (1) à la propriété correspondant à @TargetAggregateIdentifier de la commande. @Aggregate (②) est une annotation pour la liaison Spring, et a également la fonction d'annotation @Component.

Attribuez @CommandHandler (③) à la méthode qui traite la commande. La commande de création d'un nouvel agrégat est reçue par le constructeur. Cependant, le constructeur par défaut est également requis lors du traitement des commandes suivantes, car il ne rejoue l'événement qu'après instanciation.

L'événement est reçu par la méthode avec @EventSourcingHandler (④) et l'état de l'agrégat est mis à jour. Appelez AggregateLifecycle.markDeleted () à la fin de l'événement de vie Aggregate.

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 Les événements enregistrés dans le magasin d'événements sont livrés à la méthode avec @EventHandler (1). Veuillez noter qu'il est différent de @EventSourcingHandler. La requête est reçue par la méthode avec @QueryHandler (②), et les données qui remplissent les conditions sont acquises et renvoyées.

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 Il s'agit du contrôleur de Spring MVC qui traduit les requêtes HTTP en commandes et les envoie au bus de commande et au bus de requête à l'aide de la passerelle de commande et de la passerelle de requête. L'envoi de commandes à CommandGateway a une méthode d'envoi asynchrone et une méthode sendAndWait synchrone. Ici, j'utilise le modèle PRG, mais comme il peut y avoir des dépassements, j'utilise 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
  }
}

Difficulté

Je ne vois pas le résultat de la commande dans la requête

Par exemple, dans l'exemple, même si j'ai créé une application (/ factures) et redirigé pour afficher la page (/), je n'ai pas pu obtenir les données. Lorsque j'ai débogué, la méthode @CommandHandler a envoyé un événement et la méthode @EventSourcingHandler a été appelée immédiatement, mais la méthode @EventHandler était un peu plus tard. Apparemment, le processeur d'événements par défaut d'Axon, TrackingEventProcessor, s'exécute dans un thread séparé et surveille les ajouts d'événements à la valeur par défaut de 5 secondes.

C'est probablement une bonne idée d'ajuster le cycle de surveillance pour les grands systèmes, mais pour les plus petits, il est préférable de passer à un autre SubscribingEventProcessor.

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

Avec SubscribingEventProcessor, la méthode de @ EventHandler est appelée à partir du même thread que Controller, vous pouvez donc éviter le dépassement de la requête.

Impressions

Étant donné que diverses enquêtes ne sont pas encore terminées, on ne sait toujours pas si elle peut résister à un développement réel. Par exemple

• Lors de la modification de l'implémentation d'Aggregate en raison d'un changement de spécification, l'événement a déjà été enregistré, alors que dois-je faire?
• Si un bug enregistre un événement incorrect, l'événement peut-il être modifié ou supprimé?
• Je veux un historique des opérations, mais existe-t-il un mécanisme pour obtenir une liste d'événements?
• Si vous voulez assurer l'unicité entre les agrégats, comme le nom de l'application, il semble que le magasin de données s'enregistre avant d'envoyer la commande depuis le contrôleur, mais je suis un peu inquiet si cela seul peut le gérer.

Etc. J'aimerais étudier ces derniers à l'avenir.