了解 Axon 事件溯源

Question

我一直在学习轴突和事件溯源，我想我终于理解了其中的一部分，这在我的脑海中是有道理的，但我想确保我对它的理解是正确的，并且我没有做任何事情错误。代码有效，我也可以在我的 DOMAIN_EVENT_ENTRY 表中看到事件。

我将在下面发布我的代码（文档中的简单礼品卡示例）并解释我的思考过程。如果我没有正确理解它，请您帮助我以正确的方式理解该部分。

我没有包含命令/事件，因为它们非常简单，带有 id、amount 字段

首先是我的代码：

TestRunner.java

package com.example.demoaxon;

import java.util.UUID;

import org.axonframework.commandhandling.gateway.CommandGateway;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class TestRunner implements CommandLineRunner {
    
    private final CommandGateway commandGateway;
    
    @Autowired
    public TestRunner(CommandGateway commandGateway) {
        this.commandGateway = commandGateway;
    }
    
    @Override
    public void run(String... args) throws Exception {
        log.info("send command");
        String id = UUID.randomUUID().toString();
        commandGateway.sendAndWait(new IssueCardCommand(id,100));
        commandGateway.sendAndWait(new RedeemCardCommand(id,90));
        
    }
}

GiftCard.java

package com.example.demoaxon;

import org.axonframework.commandhandling.CommandHandler;
import org.axonframework.eventsourcing.EventSourcingHandler;
import org.axonframework.modelling.command.AggregateIdentifier;
import static org.axonframework.modelling.command.AggregateLifecycle.apply;
import org.axonframework.spring.stereotype.Aggregate;

import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@NoArgsConstructor
@Aggregate
@Slf4j
public class GiftCard {
    
    @AggregateIdentifier
    private String giftCardId;
    private Integer amount;
    
    @CommandHandler
    public GiftCard(IssueCardCommand cmd) {
        log.info("handling {}",cmd);
        apply(new CardIssuedEvent(cmd.getCardId(),cmd.getAmount()));
    }
    
    @EventSourcingHandler
    public void onCardIssuedEvent(CardIssuedEvent evt) {
        log.info("applying {}",evt);
        this.giftCardId = evt.getCardId();
        this.amount = evt.getAmount();
    }
    
    @CommandHandler
    public void redeemCardCommandHandler(RedeemCardCommand cmd) {
        log.info("handling {}",cmd);
        this.amount -= cmd.getAmount();
        apply(new CardRedeemedEvent(cmd.getCardId(),cmd.getTransactionId(),this.amount));
    }
    
    @EventSourcingHandler
    public void onCardRedeemedEvent(CardRedeemedEvent evt) {
        log.info("applying {}",evt);
        this.amount = evt.getAmount();
    }   
}

据我所知：

1. 在我的TestRunner 类中，命令网关使用命令总线将issueCardCommand 分派到它的CommandHandler，然后创建GiftCard 聚合的新实例。在这个CommandHandler中我们可以执行任何逻辑，然后我们使用这个apply方法。

2. apply(event) 方法用于在GiftCard 聚合范围内将CardIssuedEvent 发布为EventMessage，它还为该特定事件调用EventSourcingHandler，因此在本例中为onCardIssuedEvent .它将 EventMessage 发布到 EventBus 并发送到 EventHandlers。

3. 在@EventSourcingHandler onCardIssuedEvent 中，我们可以对GiftCard 聚合进行任何状态更改，并且我们还使用spring Jpa 将事件持久化到DOMAIN_EVENT_ENTRY 表中。

4. 一旦这个CommandHandler 执行完毕，聚合对象就不再存在了。

5. 现在再次在我的TestRunner 类中，命令网关将RedeemCardCommand 分派给它的CommandHandler，并且由于第一个命令不再存在，因此使用空的无参数构造函数来创建对象。 axon 框架从该 DOMAIN_EVENT_ENTRY 表中检索所有事件，并重放 GiftCard 聚合实例的所有事件 (EventSourcingHandlers) 以获得它的当前状态（这就是为什么 @AggregateIdentifier 很重要）。

6. 然后执行RedeemCardCommandHandler 方法，它执行任何逻辑并应用事件，该事件在聚合中发布并调用它的EventSourcingHandler。这个EventSourcingHandler 然后更新GiftCard 聚合的状态/持续到DOMAIN_EVENT_ENTRYtable。

我对事件溯源如何工作的理解正确吗？

Answer 1

让我试着引导你完成这段旅程！

您的理解几乎完全正确。我只想将其拆分为 Event Sourcing 和 Axon 以便更好地理解。

• 事件溯源

简而言之，事件溯源是一种通过过去发生的事件的历史来存储应用程序状态的方法。请记住，您还有其他模式，例如状态存储聚合。 在您的示例中，您使用的是事件溯源，因此为什么 @EventSourcingHandlers 已经到位。现在进入下一个主题。

• 轴突

我建议您阅读我们的一位同事撰写的awesome blog，特别是其中包含的幻灯片，您将看到 Axon Framework 中的消息之旅！

现在，我想先澄清一些事情：

1. 正确，您发送一个Command，并且由于注解的方法是一个构造函数，它会为您创建一个聚合。 CommandHanlders 是业务逻辑和验证的正确位置。
2. 这里apply 将在内部发布消息（发给这个Aggregate，也发给他们的Entities/AggregateMembers），然后再发给EventBus。来自 javadoc：

该事件应立即应用于聚合并安排发布到其他事件处理程序。

1. 由于我们谈论的是事件溯源，所有EventSourcingHandlers 将被调用并且Aggregate 的状态被修改/更新。这很重要，因为如前所述，这是您在需要时重建聚合状态的方式。但是这里不会发生Event的持久化，当这个过程完成时，它已经被安排好了。

2. 正确。

3. 也正确。这通常与事件溯源以及它如何重建您的Aggregate 有关。

4. 关于事件何时发布/保存/持久的第 3 点的观察结果也正确。

---

关于您的代码的另一个小注释：您在 @CommandHandler 上执行此操作

@CommandHandler
public void redeemCardCommandHandler(RedeemCardCommand cmd) {
    log.info("handling {}", cmd);
    this.amount -= cmd.getAmount(); // you should not do this here
    apply(new CardRedeemedEvent(cmd.getCardId(), cmd.getTransactionId(), this.amount));
}

此状态更改应在@EventSourcingHandler 中进行。在@CommandHandler 上，您应该仅在此聚合有足够的“钱”可以兑换时进行验证:)

Answer 2

是的，您似乎正确地使用了命令网关，并且聚合具有正确的方法并使用正确的内容进行了注释！您正在为分布式事务使用 Saga 设计模式，这比 2PC（两阶段提交）要好得多，因此调用下一步的每一步都是正确的做法。

只要开始saga方法注解@StartSaga，结束saga注解@EndSaga，那么步骤就可以按顺序进行，回滚也可以正常进行

saveAndWait 函数实际上返回一个CompletableFuture，因此，如果您愿意，可以在调试模式下单步执行线程并暂停线程，直到整个 saga 完成，如果您愿意这样做。

我唯一关心的是 - 您是使用 Axon 服务器作为事件源，还是使用 Axon 支持的其他事件总线或源？标准 Axon 服务器的唯一问题是它是 Free Tier 产品，并且他们提供了更好、更受支持的版本，例如 Axon Server Enterprise（它提供了更多功能 - 对基础架构更好） - 所以如果你这样做是为了公司，那么他们可能愿意为更高级别的额外支持付费......

Axon 框架确实支持 Spring AMQP，这很有用，因为这样，您可以更好地控制自己的基础架构，而不是被束缚在付费的 Axon Server 上——我实际上并不好或不好.

如果您可以让它与自定义事件总线/源程序一起使用，那么您的系统将非常好 - 结合在 Axon 上开发的便利性。

只要聚合具有正确的注释，并且您命令网关正确自动连接，那么一切都应该正常工作。我唯一担心的是对Axon Framework 的支持还不够...