在同一个 Spring Boot 应用程序中多次监听同一个事件时处理轴突中的异常

Question

我们正在将 axon 集成到我们现有的 Spring Boot 应用程序中。我们目前正在使用 Axon 4.1.2 和 Axon Server。

例如在注册过程中，我们触发一个RegisterCommand，它被RegisterAggregate读取并触发RegistrationDoneEvent。

有两个EventHandlers 在收听这个RegistrationDoneEvent。分别是RegistrationNeo4jEventHandler 和RegistrationSqlEventHandler。

在没有异常的情况下一切正常。但是，当出现异常时，假设 Neo4JEventHandler 收到事件，如果出现异常，则调用 SqlEventHandler sill 并且似乎所有内容仍会在 SqlEventHandler 上回滚，即使 SqlEventHandler运行成功。

我们如何才能使SqlEventHandler 完成并提交但Neo4JEventHandler 重试？

其次，我们如何在失败时完全停止重试事件？假设我们有四个事件处理程序（HandlerA、HandlerB、HandlerC、HandlerD）监听 SAME 事件。如果 HandlerC 失败，我们希望在修复基础问题时触发它重试，但还要确保其他侦听 SAME 事件的处理程序不会重新运行。

以下代码 sn-ps 包括聚合和事件处理程序。

注册聚合

@Aggregate
public class RegisterAggregate {

    ....

    @CommandHandler
    public RegisterAggregate(RegisterCommand command) {
        apply(new RegistrationDoneEvent(command));
    }
}

RegistrationSqlEventHandler

@Service
@Transactional
public class RegistrationSqlEventHandler { 

    @EventHandler
    @Order(Ordered.HIGHEST_PRECEDENCE)
    public void on(RegistrationDoneEvent event) {
        ....
    }
}

RegistrationNeo4jEventHandler

@Service
@Transactional
public class RegistrationNeo4jEventHandler { 

    @EventHandler
    public void on(RegistrationDoneEvent event) {
        ....
    }
}

Answer 1

我将把我的答案分成两部分，因为您提出的问题不是一个，而是两个。 先说说你的这个问题：

我们如何才能使RegistrationSqlEventHandler 完成并提交但RegistrationNeo4jEventHandler 重试？

从命名中可以看出，两个事件处理组件（即 you 编写的包含 @EventHandler 注释方法的类）服务于一个完全不同的查询模型。第一个是 RDBMS，第二个是通过 Neo4j 更新 Graph Model。

因此，我发现您最终很可能对两者都有不同的非功能性要求。为了能够允许对这些进行不同的配置，您必须为两者使用不同的 Event Processor 实例。事件处理器，负责管理向您的事件处理程序提供事件的技术方面的组件，作为配置异常处理、线程数、批处理大小等的地方。

请注意，事件处理器有两种形式：SubscribingEventProcessor 和 TrackingEventProcessor。这些可以简单地描述为事件推送和事件拉取机制，后者是默认设置，因为它强制进一步隔离。

要为两者配置不同的事件处理器，您可以使用 Axon 提供的配置 API。对于事件处理器，这意味着与EventProcessingConfigurer 进行交互。使用它，您可以定义不同的事件处理器，然后将您的事件处理组件分配给正确的实例。您可以使用的简写方法是在两个事件处理组件上添加 @ProcessingGroup 注释，其中包含不同的名称。 尤其是如果您在 Spring Boot 环境中，就配置而言，这已经足够了。

进行这种隔离将确保RegistrationNeo4jEventHandler 中的异常情况不会对RegistrationSqlEventHandler 产生任何不良影响，反之亦然。

---

其次，让我们转到您的另一个问题：

其次，我们如何在失败时完全停止重试事件？

为此，您必须调整事件处理过程的异常处理。当涉及到事件处理组件时，Axon 派生了两个级别的异常处理：

1. ListenerInvocationErrorHandler -> 负责处理@EventHandler 注解的方法抛出的异常。
2. ErrorHandler -> 负责处理事件处理器内抛出的异常。

这些的默认实现将分别记录错误（使用LoggingErrorHandler）并传播异常（使用PropagatingErrorHandler）。

仅供参考，参考指南中有 this 来说明事件处理器的错误处理。

在这个问题中，您将使用不同的事件处理功能进一步指定您的案例。同样，如果您不想影响一个事件处理的失败而导致另一个事件处理出现问题，您可能希望将这些问题与不同的事件处理器隔离开来。

请注意，如果您所示例的这四个事件处理程序只是更新查询模型，那么后续调用应该只是执行相同的操作而没有任何副作用。但是，如果这些事件处理程序执行一些其他（外部）活动，例如发送电子邮件，那么肯定需要将给定的事件处理组件隔离到一个您不想重试/重播的不同事件处理器中。