Axon框架指南 - Baeldung


在本文中，我们将介绍Axon以及它如何帮助我们实现具有CQRS（Command Query Responsibility Segregation）和Event Sourcing的应用程序。
在本指南中，将使用Axon Framework和Axon Server。前者将包含我们的实现，后者将是我们专用的事件存储和消息路由解决方案。
我们将要构建的示例应用程序专注于Order域。为此，我们将利用Axon为我们提供的CQRS和Event Sourcing构建模块。
请注意，很多共享概念都来自DDD，这超出了本文的范围。

Maven依赖
我们将创建一个Axon / Spring Boot应用程序。因此，我们需要将最新的axon-spring-boot-starter依赖项添加到我们的pom.xml中，以及用于测试的axon-test依赖项：

<dependency>
    <groupId>org.axonframework</groupId>
    <artifactId>axon-spring-boot-starter</artifactId>
    <version>4.1.2</version>
</dependency>
 
<dependency>
    <groupId>org.axonframework</groupId>
    <artifactId>axon-test</artifactId>
    <version>4.1.2</version>
    <scope>test</scope>
</dependency>

 Axon Server
我们将使用Axon Server作为我们的Event Store和我们的专用命令，事件和查询路由解决方案。
作为事件存储，它为我们提供了存储事件时所需的理想特性。该文章提供了背景为什么这是可取的。
作为消息路由解决方案，它为我们提供了将多个实例连接在一起的选项，而无需专注于配置RabbitMQ或Kafka主题以共享和分发消息。
Axon Server可以在这里下载。由于它是一个简单的JAR文件，以下操作足以启动它：

java -jar axonserver.jar

这将启动一个可通过localhost访问的Axon Server实例：8024。端点提供已连接应用程序及其可以处理的消息的概述，以及Axon Server中包含的事件存储的查询机制。
Axon Server的默认配置与axon-spring-boot-starter依赖关系将确保我们的Order服务将自动连接到它。

订单服务API - 命令
我们将以CQRS为基础设置订单服务。因此，我们将强调流经我们应用程序的消息。
首先，我们将定义命令，即意图的表达。Order服务能够处理三种不同类型的操作：

  1. 下新订单
  2. 确认订单
  3. 发货订单

当然，我们的域可以处理三个命令消息 -  PlaceOrderCommand，ConfirmOrderCommand和ShipOrderCommand：

public class PlaceOrderCommand {
  
    @TargetAggregateIdentifier
    private final String orderId;
    private final String product;
  
    // constructor, getters, equals/hashCode and toString 
}
public class ConfirmOrderCommand {
  
    @TargetAggregateIdentifier
    private final String orderId;
     
    // constructor, getters, equals/hashCode and toString
}
public class ShipOrderCommand {
  
    @TargetAggregateIdentifier
    private final String orderId;
  
    // constructor, getters, equals/hashCode and toString
}

该TargetAggregateIdentifier注解告诉轴突的注释字段是一个给定的聚合ID，以该命令应该有针对性。 我们将在本文后面简要介绍聚合。
另请注意，我们将命令中的字段标记为  final。 这是故意的，因为任何消息实现都是不可变的最佳实践。

订单服务API - 事件
我们的聚合将处理这些命令，因为它负责决定是否可以下达，确认或发送订单。
它将通过发布活动通知其决定的其余部分。我们将有三种类型的事件 -  OrderPlacedEvent，OrderConfirmedEvent和OrderShippedEvent：

public class OrderPlacedEvent {
  
    private final String orderId;
    private final String product;
  
    // default constructor, getters, equals/hashCode and toString
}
public class OrderConfirmedEvent {
  
    private final String orderId;
  
    // default constructor, getters, equals/hashCode and toString
}
public class OrderShippedEvent { 
 
    private final String orderId; 
 
    // default constructor, getters, equals/hashCode and toString 
}

命令模型 - 订单聚合
现在我们已经根据命令和事件建模了我们的核心API，我们可以开始创建命令模型。
由于我们的领域专注于处理订单，  我们将创建一个OrderAggregate作为我们的命令模型的中心。

聚合类，创建我们的基本聚合类：

@Aggregate
public class OrderAggregate {
 
    @AggregateIdentifier
    private String orderId;
    private boolean orderConfirmed;
 
    @CommandHandler
    public OrderAggregate(PlaceOrderCommand command) {
        AggregateLifecycle.apply(new OrderPlacedEvent(command.getOrderId(), command.getProduct()));
    }
 
    @EventSourcingHandler
    public void on(OrderPlacedEvent event) {
        this.orderId = event.getOrderId();
        orderConfirmed = false;
    }
 
    protected OrderAggregate() { }
}

使用@Aggregate注释标记这个类作为一个聚合体。它将通知框架需要为此OrderAggregate实例化所需的CQRS和Event Sourcing特定构建块。
由于聚合将处理针对特定聚合实例的命令，因此我们需要使用AggregateIdentifier注释指定标识符。
在OrderAggregate '命令处理构造函数'中处理PlaceOrderCommand时，我们的聚合将开始其生命周期。为了告诉框架使用指定函数处理命令，我们将添加CommandHandler注释。
处理PlaceOrderCommand时，它将通过发布OrderPlacedEvent通知应用程序的其余部分已下达订单。要从聚合中发布事件，我们将使用  AggregateLifecycle application（Object ...）。
从这一点开始，我们实际上可以开始将Event Sourcing作为从事件流中重新创建聚合实例的驱动力。
我们从“聚合创建事件”开始，即OrderPlacedEvent，它在EventSourcingHandler注释函数中处理，以设置Order聚合的orderId和orderConfirmed状态。
另请注意，为了能够根据事件来源聚合，Axon需要一个默认构造函数。

聚合命令处理程序
现在我们有了基本聚合，我们可以开始实现剩余的命令处理程序：

@CommandHandler
public void handle(ConfirmOrderCommand command) { 
    apply(new OrderConfirmedEvent(orderId)); 
} 
 
@CommandHandler
public void handle(ShipOrderCommand command) { 
    if (!orderConfirmed) { 
        throw new UnconfirmedOrderException(); 
    } 
    apply(new OrderShippedEvent(orderId)); 
} 
 
@EventSourcingHandler
public void on(OrderConfirmedEvent event) { 
    orderConfirmed = true; 
}

我们已经定义订单只有在确认后才能发货。因此，如果不是这种情况，我们将抛出UnconfirmedOrderException。
这表明OrderConfirmedEvent采购处理程序需要将Order聚合的orderConfirmed状态更新为true。

测试命令模型
首先，我们需要创建一个为OrderAggregate测试的配置FixtureConfiguration ：

private FixtureConfiguration<OrderAggregate> fixture;
 
@Before
public void setUp() {
    fixture = new AggregateTestFixture<>(OrderAggregate.class);
}

第一个测试用例应该涵盖最简单的情况。当聚合处理  PlaceOrderCommand时，它应该生成一个  OrderPlacedEvent：

String orderId = UUID.randomUUID().toString();
String product = "Deluxe Chair";
fixture.givenNoPriorActivity()
  .when(new PlaceOrderCommand(orderId, product))
  .expectEvents(new OrderPlacedEvent(orderId, product));

接下来，我们可以测试只有在确认后能够发送订单的决策逻辑。因此，我们有两个场景 - 一个是我们期望异常的场景，另一个是我们期望  OrderShippedEvent的场景。
让我们看看第一个场景，我们期待一个异常：

String orderId = UUID.randomUUID().toString();
String product = "Deluxe Chair";
fixture.given(new OrderPlacedEvent(orderId, product))
  .when(new ShipOrderCommand(orderId))
  .expectException(IllegalStateException.class);

现在是第二种情况，我们期待OrderShippedEvent：

String orderId = UUID.randomUUID().toString();
String product = "Deluxe Chair";
fixture.given(new OrderPlacedEvent(orderId, product), new OrderConfirmedEvent(orderId))
  .when(new ShipOrderCommand(orderId))
  .expectEvents(new OrderShippedEvent(orderId));

查询模型 - 事件处理程序
到目前为止，我们已经使用命令和事件建立了我们的核心API，并且我们拥有CQRS Order服务的Command模型，Order aggregate。
接下来，  我们可以开始考虑我们的应用程序应该服务的查询模型之一。
其中一个模型是OrderedProducts：

public class OrderedProduct {
 
    private final String orderId;
    private final String product;
    private OrderStatus orderStatus;
 
    public OrderedProduct(String orderId, String product) {
        this.orderId = orderId;
        this.product = product;
        orderStatus = OrderStatus.PLACED;
    }
 
    public void setOrderConfirmed() {
        this.orderStatus = OrderStatus.CONFIRMED;
    }
 
    public void setOrderShipped() {
        this.orderStatus = OrderStatus.SHIPPED;
    }
 
    // getters, equals/hashCode and toString functions
}
public enum OrderStatus {
    PLACED, CONFIRMED, SHIPPED
}

我们将根据通过系统传播的事件更新此模型。用于更新模型的Spring Service bean可以解决这个问题：

@Service
public class OrderedProductsEventHandler {
 
    private final Map<String, OrderedProduct> orderedProducts = new HashMap<>();
 
    @EventHandler
    public void on(OrderPlacedEvent event) {
        String orderId = event.getOrderId();
        orderedProducts.put(orderId, new OrderedProduct(orderId, event.getProduct()));
    }
 
    // Event Handlers for OrderConfirmedEvent and OrderShippedEvent...
}

由于我们已经使用axon-spring-boot-starter依赖来启动我们的Axon应用程序，因此框架将自动扫描所有bean以查找现有的消息处理函数。
由于  OrderedProductsEventHandler具有用于存储OrderedProduct并更新它的EventHandler注释函数，因此该bean将被框架注册为应该接收事件而不需要我们任何配置的类。

查询模型 - 查询处理程序
接下来，要查询此模型，例如，要检索所有已订购的产品，我们应首先向我们的核心API引入一条Query消息：
public class FindAllOrderedProductsQuery { }

其次，我们必须更新OrderedProductsEventHandler才能处理FindAllOrderedProductsQuery：

@QueryHandler
public List<OrderedProduct> handle(FindAllOrderedProductsQuery query) {
    return new ArrayList<>(orderedProducts.values());
}

该QueryHandler注释功能将处理FindAllOrderedProductsQuery并设置为返回一个List<OrderedProduct>，类似“find all”查询。

把所有东西放在一起
我们通过命令，事件和查询充实了我们的核心API，并通过OrderAggregate和OrderedProducts模型设置了我们的命令和查询模型。
接下来是绑定我们基础设施的松散端。当我们使用axon-spring-boot-starter时，它会自动设置许多所需的配置。
首先，由于我们想要为我们的聚合利用事件采购，我们需要一个EventStore。我们在第三步中启动的Axon Server将填补这个漏洞。
其次，我们需要一种机制来存储我们的OrderedProduct查询模型。对于此示例，我们可以添加h2作为内存数据库和spring-boot-starter-data-jpa以便于使用：

<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>

设置REST端点
接下来，我们需要能够访问我们的应用程序，我们将通过添加spring-boot-starter-web依赖关系来利用REST端点：

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

从我们的REST端点，我们可以开始调度命令和查询：

@RestController
public class OrderRestEndpoint {
 
    private final CommandGateway commandGateway;
    private final QueryGateway queryGateway;
 
    // Autowiring constructor and POST/GET endpoints
}

该CommandGateway被用作机制发送我们的命令消息，以及QueryGateway，然后发送查询消息，
与 CommandBus和QueryBus相比，该网关提供了更简单，更直接的API 。
从这里开始，我们的OrderRestEndpoint应该有一个POST端点来放置，确认和发送订单：

@PostMapping("/ship-order")
public void shipOrder() {
    String orderId = UUID.randomUUID().toString();
    commandGateway.send(new PlaceOrderCommand(orderId, "Deluxe Chair"));
    commandGateway.send(new ConfirmOrderCommand(orderId));
    commandGateway.send(new ShipOrderCommand(orderId));
}

这使我们的CQRS应用程序的命令端更加完整。
现在，剩下的就是一个GET端点来查询所有OrderedProducts：
@GetMapping("/all-orders")
public List<OrderedProduct> findAllOrderedProducts() {
    return queryGateway.query(new FindAllOrderedProductsQuery(), 
      ResponseTypes.multipleInstancesOf(OrderedProduct.class)).join();
}

在GET端点中，我们利用QueryGateway来分派点对点查询。于是，我们创建一个默认的  FindAllOrderedProductsQuery，但我们还需要指定预期的返回类型。
由于我们期望返回多个OrderedProduct实例，因此我们利用静态ResponseTypes＃multipleInstancesOf（Class）函数。有了这个，我们为订单服务的查询方面提供了一个基本入口。

我们完成了设置，现在我们可以在启动OrderApplication后通过REST控制器发送一些命令和查询  。
POST到端点/发货订单将实例化一个OrderAggregate，它将发布事件，这反过来将保存/更新我们的OrderedProducts。来自/ all-orders  端点的GET 将发布一个查询消息，该消息将由OrderedProductsEventHandler处理，该消息将返回所有现有的OrderedProducts。

结论
在本文中，我们介绍了Axon Framework作为构建应用程序的强大基础，充分利用了CQRS和Event Sourcing的优势。
我们使用框架实现了一个简单的Order服务，以展示如何在实践中构建这样的应用程序。
最后，Axon Server构成了我们的事件存储和消息路由机制。
可以在GitHub上找到所有这些示例和代码片段的实现。
如果您有任何其他问题，请查看Axon Framework用户组。

百度搜索提示网站被篡改了，版主！