Axon：创建聚合后在 Saga 中创建并保存另一个聚合

Question

更新： 问题似乎是我使用了两次的 id，或者换句话说，我想用于 productinventory 的产品实体的 id实体。一旦我为 productinventory 实体生成一个新的 id，它似乎工作正常。但我希望两者具有相同的 ID，因为它们是相同的产品。

我有 2 个服务：

ProductManagementService（保存包含产品详细信息的 Product 实体）

1.) 为了保存产品实体，我实现了一个监听 ProductCreatedEvent 并将产品保存到 mysql 数据库的事件处理程序。

ProductInventoryService（将具有产品库存数量的 ProductInventory 实体保存到 ProductManagementService 中定义的某个 productId）

2.) 为了保存 ProductInventory 实体，我还实现了一个监听 ProductInventoryCreatedEvent 并将产品保存到 mysql 数据库的事件处理程序。

我想做的事情：

当在 ProductManagementService 中创建新产品时，我想随后直接在 ProductInventoryService 中创建一个 ProductInventory 实体并将其保存到我的 msql table。新的 ProductInventory 实体应与 Product 实体具有相同的 ID。

为此，我创建了一个 Saga，它列出了 ProductCreatedEvent 并发送了一个新的 CreateProductInventoryCommand。一旦 CreateProductInventoryCommand 触发 ProductInventoryCreatedEvent，2.) 中描述的 EventHandler 应该捕获它。除了它没有。

唯一保存的是产品实体，所以总而言之：

1.) 有效，2.) 无效。确实创建了 ProductInventory 聚合，但未保存它，因为未触发连接到 EventHandler 的保存过程。

我也遇到异常，但应用程序没有崩溃：Command 'com.myApplication.apicore.command.CreateProductInventoryCommand' resulted in org.axonframework.commandhandling.CommandExecutionException(OUT_OF_RANGE: [AXONIQ-2000] Invalid sequence number 0 for aggregate 3cd71e21-3720-403b-9182-130d61760117, expected 1)

我的传奇：

@Saga
@ProcessingGroup("ProductCreationSaga")
public class ProductCreationSaga {

    @Autowired
    private transient CommandGateway commandGateway;

    @StartSaga
    @SagaEventHandler(associationProperty = "productId")
    public void handle(ProductCreatedEvent event) {
        System.out.println("ProductCreationSaga, SagaEventHandler, ProductCreatedEvent");

        String productInventoryId = event.productId;
        SagaLifecycle.associateWith("productInventoryId", productInventoryId);
    //takes ID from product entity and sets all 3 stock attributes to zero
        commandGateway.send(new CreateProductInventoryCommand(productInventoryId, 0, 0, 0));
    }

    @SagaEventHandler(associationProperty = "productInventoryId")
    public void handle(ProductInventoryCreatedEvent event) {
        System.out.println("ProductCreationSaga, SagaEventHandler, ProductInventoryCreatedEvent");

        SagaLifecycle.end();
    }

}

按预期工作并保存产品实体的事件处理程序：

@Component
public class ProductPersistenceService {

    @Autowired
    private ProductEntityRepository productRepository;

    //works as intended
    @EventHandler
    void on(ProductCreatedEvent event) {
        System.out.println("ProductPersistenceService, EventHandler, ProductCreatedEvent");
        ProductEntity entity = new ProductEntity(event.productId, event.productName, event.productDescription, event.productPrice);
        productRepository.save(entity);

    }

    @EventHandler
    void on(ProductNameChangedEvent event) {
        System.out.println("ProductPersistenceService, EventHandler, ProductNameChangedEvent");
        ProductEntity existingEntity = productRepository.findById(event.productId).get();

        ProductEntity entity = new ProductEntity(event.productId, event.productName, existingEntity.getProductDescription(), existingEntity.getProductPrice());
        productRepository.save(entity);

    }
}

应该保存 ProductInventory 实体但没有保存的事件处理程序：

@Component
public class ProductInventoryPersistenceService {

    @Autowired
    private ProductInventoryEntityRepository productInventoryRepository;

    //doesn't work
    @EventHandler
    void on(ProductInventoryCreatedEvent event) {
        System.out.println("ProductInventoryPersistenceService, EventHandler, ProductInventoryCreatedEvent");
        ProductInventoryEntity entity = new ProductInventoryEntity(event.productInventoryId, event.physicalStock, event.reservedStock, event.availableStock);
        System.out.println(entity.toString());

        productInventoryRepository.save(entity);

    }

}

产品汇总：

@Aggregate
public class Product {

    @AggregateIdentifier
    private String productId;

    private String productName;

    private String productDescription;

    private double productPrice;

    public Product() {
    }

    @CommandHandler
    public Product(CreateProductCommand command) {
        System.out.println("Product, CommandHandler, CreateProductCommand");
        AggregateLifecycle.apply(new ProductCreatedEvent(command.productId, command.productName, command.productDescription, command.productPrice));
    }

    @EventSourcingHandler
    protected void on(ProductCreatedEvent event) {
        System.out.println("Product, EventSourcingHandler, ProductCreatedEvent");

        this.productId = event.productId;
        this.productName = event.productName;
        this.productDescription = event.productDescription;
        this.productPrice = event.productPrice;
    }

}

产品库存汇总：

@Aggregate
public class ProductInventory {

    @AggregateIdentifier
    private String productInventoryId;

    private int physicalStock;

    private int reservedStock;

    private int availableStock;

    public ProductInventory() {
    }


    @CommandHandler
    public ProductInventory(CreateProductInventoryCommand command) {
        System.out.println("ProductInventory, CommandHandler, CreateProductInventoryCommand");
        AggregateLifecycle.apply(new ProductInventoryCreatedEvent(command.productInventoryId, command.physicalStock, command.reservedStock, command.availableStock));
    }

    @EventSourcingHandler
    protected void on(ProductInventoryCreatedEvent event) {
        System.out.println("ProductInventory, EventSourcingHandler, ProductInventoryCreatedEvent");

        this.productInventoryId = event.productInventoryId;
        this.physicalStock = event.physicalStock;
        this.reservedStock = event.reservedStock;
        this.availableStock = event.availableStock;
    }

}

Answer 1

您现在注意到的是给定事件存储中 [聚合标识符、序列号] 对的唯一性要求。此要求旨在保护您免受对同一聚合实例的潜在并发访问，因为同一聚合的多个事件都需要具有唯一的整体序列号。此数字还用于标识需要处理事件的顺序，以保证以相同的顺序一致地重新创建聚合。

因此，您可能会认为这会选择“抱歉，没有合适的解决方案”，但幸运的是情况并非如此。您可以在此设置中大致做三件事：

事实上，两个聚合都将具有唯一标识符。
在两个应用程序之间使用不同的限界上下文。
更改聚合标识符的书写方式。

选项 1 可以说是最实用的并且被大多数人使用。但是，您已经注意到标识符的重用是必要的，所以我假设您已经完全忽略了这个选项。无论如何，我会尝试重新审视这种方法，因为对您创建的每个新实体默认使用 UUIDs 可以避免您将来遇到麻烦。

选项 2 将通过 DDD. Letting the Product aggregate and ProductInventory aggregate reside in distinct contexts will mean you will have distinct event stores for both. Thus, the uniqueness constraint would be kept, as no single store is containing both aggregate event streams. Whether this approach is feasible however depends on whether both aggregates actually belong to the same context yes/no. If this is the case, you could for example use Axon Server's multi-context support 引入的限界上下文概念来反映自身，以创建两个不同的应用程序。

选项 3 需要对 Axon 的功能有一点了解。当它存储事件时，它将在聚合中的 @AggregateIdentifier 注释字段上调用 toString() 方法。由于您的 @AggregateIdentifier 注释字段是 String，您将按原样获得标识符。您可以做的是输入标识符，toString() 方法不只是 return 标识符，而是将聚合类型附加到它。这样做会使存储的 aggregateIdentifier 独一无二，而从使用的角度来看，您似乎仍然在重复使用标识符。

这三个选项中的哪一个更适合您的解决方案，从我的角度来看很难推断出来。我所做的是，从我的角度来看，以最合理的方式订购它们。希望这会帮助你进一步@Jan!

Axon：创建聚合后在 Saga 中创建并保存另一个聚合

Axon: Create and Save another Aggregate in Saga after creation of an Aggregate

java

spring-boot

event-handling

event-sourcing

axon