如何在Axon中实现事件重放？跨节点微服务场景的可行方案

1. 基础场景下的Axon事件重放实现 #

Axon核心是通过**Tracking Event Processor (TEP)**来处理事件重放的，这是最常用的落地方式，具体步骤如下：

• 步骤1：确认事件处理器类型
  Axon默认的事件处理器就是Tracking类型，如果需要自定义配置，可以通过配置类调整：

  @Configuration
  public class AxonConfig {
      @Bean
      public EventProcessorConfiguration eventProcessorConfiguration() {
          return EventProcessorConfiguration.forTrackingEventProcessor()
                  // 重放时调大批量处理数，提升效率
                  .batchSize(50);
      }
  }
  

• 步骤2：重置Tracking Token触发重放
  要让处理器重新消费历史事件，需要重置它的tracking token，让它回到事件流的起始点（或指定位置）。你可以通过API手动触发：

  @Autowired
  private EventProcessor orderEventProcessor;
  
  public void triggerReplay() {
      if (orderEventProcessor instanceof TrackingEventProcessor tep) {
          // 重置到事件流最开始
          tep.resetTokens();
          // 也可以指定重置到某个时间点
          // tep.resetTokens(Instant.parse("2024-01-01T00:00:00Z"));
          // 确保处理器处于运行状态
          tep.start();
      }
  }
  

• 步骤3：区分重放与实时事件（可选）
  如果重放时需要清空旧投影再重建，可以通过ReplayStatus参数判断当前事件是否来自重放：

  @ProcessingGroup("order-processor")
  @Component
  public class OrderProjection {
      @EventHandler
      public void handle(OrderCreatedEvent event, ReplayStatus replayStatus) {
          if (replayStatus.isReplay()) {
              // 重放时先删除旧数据，避免重复
              orderRepository.deleteById(event.getOrderId());
          }
          orderRepository.save(new OrderProjectionModel(event.getOrderId(), event.getStatus()));
      }
  }
  

2. 跨微服务（命令端+查询端）场景下的事件重放方案 #

你的场景是命令端用Mongo存事件、RabbitMQ发事件，查询端订阅RabbitMQ构建投影。这种跨节点独立部署的情况，默认的Axon本地重放机制无法直接生效——因为查询端没有自己的事件存储，仅依赖RabbitMQ的实时推送，而历史事件只存在命令端的Mongo里。下面是几种可行的解决方案：

方案一：共享命令端的Event Store（快速实现但有耦合） #

让查询端的Axon配置直接连接命令端的Mongo Event Store，这样查询端的Tracking Processor就能直接读取历史事件进行重放：

• 在查询端的application.properties中配置：

  axon.eventstore.mongo.database=command-side-event-db
  axon.eventstore.mongo.connection-string=mongodb://command-side-mongo:27017
  

• 之后就可以按照基础场景的步骤，重置查询端的Tracking Processor完成重放。

注意：这种方案会让查询端依赖命令端的数据库，违反微服务隔离原则，适合小型项目或临时应急场景。

方案二：命令端提供事件导出接口，查询端批量拉取重建 #

这种方案更符合微服务的隔离设计：

1. 命令端开发事件导出接口：通过Axon的EventStore API，支持按分页或时间范围导出历史事件：

   @RestController
   @RequestMapping("/events")
   public class EventExportController {
       @Autowired
       private EventStore eventStore;
   
       @GetMapping("/export")
       public List<DomainEventMessage<?>> exportEvents(@RequestParam int page, @RequestParam int size) {
           return eventStore.readEvents("")
                   .asStream()
                   .skip((long) page * size)
                   .limit(size)
                   .collect(Collectors.toList());
       }
   }
   

2. 查询端实现批量重建逻辑：先清空现有投影数据，再分页调用命令端接口拉取事件，逐个触发投影处理：

   @Service
   public class ProjectionRebuildService {
       @Autowired
       private RestTemplate restTemplate;
       @Autowired
       private OrderProjection orderProjection;
       @Autowired
       private OrderRepository orderRepository;
   
       public void rebuildOrderProjection() {
           // 先清空旧投影
           orderRepository.deleteAll();
           // 分页拉取所有事件
           int page = 0;
           while (true) {
               List<DomainEventMessage<?>> events = restTemplate.getForObject(
                       "http://command-service/events/export?page={page}&size=100",
                       List.class, page);
               if (events.isEmpty()) break;
               // 逐个处理重放事件
               events.forEach(event -> orderProjection.handle((OrderCreatedEvent) event.getPayload(), ReplayStatus.REPLAY));
               page++;
           }
       }
   }
   

方案三：引入Axon Server（推荐的长期方案） #

如果项目有长期演进需求，替换RabbitMQ为Axon Server是最优解：

• Axon Server自带分布式事件存储和事件总线，命令端的事件会自动存储到Axon Server，查询端可以直接从Axon Server订阅事件（包括历史事件重放）。
• 重放时只需要在查询端重置Tracking Event Processor的token，Axon Server会自动把历史事件推送给查询端，完全不需要手动处理跨服务的事件同步。
• 这种方案遵循Axon的最佳实践，既保持微服务隔离性，还提供事件追踪、监控等额外能力。

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