Axon应用重启后事件处理器重复读取事件 如何提交消费进度

你的重复消费问题核心是TrackingEventProcessor(TEP)的消费进度token没有被持久化成功，每次重启应用找不到上一次的消费进度，就会触发初始token配置逻辑重新消费。结合你给出的配置，按以下步骤排查修复即可：

1. 优先排查JpaTokenStore生效问题 #

你已经配置了JpaTokenStore，但大概率是TokenEntry实体没有被你的JPA框架扫描到，导致消费进度根本没有存到数据库：

• 检查你的Spring Boot启动类的@EntityScan注解，必须包含Axon的TokenEntry所在包路径，修改示例如下：

@EntityScan(basePackages = {"你的项目实体包路径", "org.axonframework.eventhandling.tokenstore.jpa"})
@SpringBootApplication
public class YourApplication {
    public static void main(String[] args) {
        SpringApplication.run(YourApplication.class, args);
    }
}

• 启动应用后检查数据库是否自动生成了token_entry表，如果表存在，消费过事件后表里会有对应事件处理器的进度记录。

2. 修正TrackingEventProcessor初始token配置逻辑 #

你当前配置的andInitialTrackingToken(StreamableMessageSource::createHeadToken)是没有已存储的token时，从事件流的最新位置开始消费，不会消费重启前的旧事件，如果你需要首次启动时消费所有历史事件，可以把初始token改成createTailToken，但注意这个配置仅在首次启动、没有token记录时生效，后续重启都会优先读已存储的token。
如果你希望自定义不同事件处理器的初始策略，可以单独给指定处理器配置，不需要全局统一配置：

eventProcessingConfigurer.registerTrackingEventProcessor(
    "你的处理器名称",
    org.axonframework.config.Configuration::eventStore,
    c -> TrackingEventProcessorConfiguration.forSingleThreadedProcessing()
            .andInitialTrackingToken(StreamableMessageSource::createTailToken)
);

3. 消费进度提交规则调整 #

Axon的TEP默认会批量消费完一批事件后自动提交token，不需要你手动提交，如果你要调整提交频率，可以在配置文件中修改：

axon:
  eventhandling:
    processors:
      # 全局配置所有TEP
      "[default]":
        mode: tracking
        batch-size: 10 # 每消费10条事件提交一次进度
        token-store: jpaTokenStore # 明确指定用JpaTokenStore存进度

如果你的业务要求每消费一条就立即提交，可以给事件处理方法加@CommitToken注解：

@EventHandler
@CommitToken
public void on(YourEvent event) {
    // 你的业务处理逻辑
}

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