Axon框架事件处理器异常处理方案咨询

首先纠正你当前代码的问题：手动从聚合中提取事件并调用eventBus.publish()的方式不符合Axon设计规范，会破坏框架自带的事务一致性机制。正确做法是在聚合的业务方法中使用apply()方法发布事件，Axon会自动管理事件的存储与发布流程。

修改你的ChatRoom聚合类：

public class ChatRoom {
    // 聚合标识、其他字段...

    public ChatRoom(String roomId, String name) {
        this.roomId = roomId;
        this.name = name;
    }

    public void create() {
        // 使用apply发布领域事件，Axon会暂存该事件
        apply(new RoomCreatedEvent(this.roomId, this.name));
    }
}

对应的命令处理器可简化为：

@CommandHandler
public void createRoom(CreateRoomCommand createRoomCommand){
    log.info("Receiving create room command: {}",createRoomCommand);
    ChatRoom chatRoom = new ChatRoom(
            createRoomCommand.getRoomId(),
            createRoomCommand.getName()
    );

    chatRoom.create();
    // Axon会在保存聚合时，将暂存的事件存入EventStore，且该操作与聚合保存处于同一事务
    chatRoomRepository.save(chatRoom);
}

接下来针对你的两个问题逐一解答：

1. 是否应在事件处理器中分发事件以回滚事务？ #

不行。默认情况下，事件处理器在命令对应的事务提交之后执行（异步模式），此时聚合已持久化到数据库，事务已完成，无法回滚。即使在事件处理器中发布回滚事件，也无法撤销已提交的数据库操作，反而会引入额外复杂度和数据不一致风险。

2. 如何让事件处理与命令事务绑定，异常时一起失败？ #

如果希望事件处理与命令事务绑定，异常时触发整体回滚，推荐使用Axon的SUBSCRIBING模式事件处理器，而非手动控制eventBus.publish：

配置同步事件处理器 #

1. 在事件处理器类上指定处理组：

@ProcessingGroup("sync-chat-group")
@Component
public class ChatRoomEventHandler {
    @EventHandler
    public void on(RoomCreatedEvent event){
        // 业务逻辑，抛出异常会触发事务回滚
    }
}

2. 在配置类中设置该处理组为SUBSCRIBING模式：

@Configuration
public class AxonConfig {
    @Autowired
    public void configureSyncEventProcessing(EventProcessingConfigurer configurer) {
        configurer.registerProcessingGroup("sync-chat-group", config ->
            // 设置为SUBSCRIBING模式，事件处理器会在命令事务内同步执行
            config.processingMode(ProcessingMode.SUBSCRIBING)
        );
    }
}

这种模式下，事件处理器与命令处理器处于同一事务上下文，一旦事件处理器抛出异常，整个事务会回滚，聚合的保存操作也会被撤销。

异步场景下的异常处理 #

如果业务场景适合异步处理事件（比如耗时操作），无法使用SUBSCRIBING模式，可通过Axon的错误处理机制应对异常：

• 配置重试策略：失败后自动重试几次，避免临时故障导致的失败
• 配置死信队列：多次重试失败后，将事件放入死信队列，不阻塞其他事件处理，后续可手动排查处理

示例配置：

@Autowired
public void configureAsyncEventProcessing(EventProcessingConfigurer configurer) {
    configurer.registerProcessingGroup("async-chat-group", config ->
        config.processingMode(ProcessingMode.TRACKING)
              // 配置死信队列错误处理器
              .errorHandler(configuration -> new DeadLetterQueueErrorHandler())
              // 配置重试策略：最多重试3次，间隔1秒
              .retryScheduler(configuration ->
                  IntervalRetryScheduler.builder()
                      .maxRetryCount(3)
                      .retryInterval(Duration.ofSeconds(1))
                      .build()
              )
    );
}

内容的提问来源于stack exchange，提问作者fragilepriCe