Axon框架指南 - Baeldung

19-07-24 banq
                   

在本文中,我们将介绍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,然后发送查询消息,

与 CommandBusQueryBus相比,该网关提供了更简单,更直接的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用户组

                   

2
abcaxyzx
2019-07-24 15:05

百度搜索提示网站被篡改了,版主!