Apache Camel对接ActiveMQ：断连消息丢失的官方优化方案咨询

完全不用自己写数据库定时器那套复杂逻辑，Camel内置了多种机制可以组合实现你的需求——断连时自动暂存消息，连接恢复后优先发送缓存消息，同时保证顺序和避免重复投递：

1. 异常路由+本地持久化缓存 #

通过OnException捕获ActiveMQ连接异常，将失败消息转存到本地持久化存储（比如文件系统），再单独监听缓存目录做补发：

• 主路由配置：捕获JMS连接异常，将消息写入本地缓存，同时记录日志

  onException(JMSException.class)
      .handled(true)
      // 用时间戳命名文件保证消息顺序
      .to("file:./activemq-fallback-cache?fileName=msg-${date:now:yyyyMMddHHmmssSSS}.json")
      .log(LoggingLevel.WARN, "ActiveMQ连接中断，消息暂存到本地缓存");
  
  // 你的业务主路由
  from("your-business-source")
      .to("activemq:queue:target-queue?usePersistentMessages=true");
  

• 缓存补发路由：监听缓存目录，按文件名（时间戳）排序发送，发送成功后删除文件避免重复

  from("file:./activemq-fallback-cache?delete=true&sortBy=file:name")
      .log("补发本地缓存消息到ActiveMQ")
      .to("activemq:queue:target-queue");
  

  这里sortBy=file:name严格保证消息按存入顺序发送，delete=true确保消息只被发送一次。

2. 幂等消费器防重复 #

如果担心缓存消息在补发时出现重复（比如路由重启），可以给补发路由添加幂等消费器，基于消息唯一标识去重：

from("file:./activemq-fallback-cache?delete=true&sortBy=file:name")
    // 用消息的唯一ID作为幂等键，这里假设消息头存在msgId字段
    .idempotentConsumer(header("msgId"), JdbcIdempotentRepository.create(jdbcTemplate, "msg_idempotent_table"))
    .to("activemq:queue:target-queue");

如果不需要持久化的幂等存储，换成MemoryIdempotentRepository即可（适合单机场景）。

3. 熔断组件优化重试逻辑 #

结合Resilience4j熔断组件，当ActiveMQ连接失败率达到阈值时，直接切换到本地缓存模式，避免无效的重试消耗资源：

from("your-business-source")
    .circuitBreaker()
        .resilience4jConfiguration()
            .failureRateThreshold(50) // 失败率50%时触发熔断
            .waitDurationInOpenState(10000) // 熔断10秒后尝试恢复连接
            .build()
        .to("activemq:queue:target-queue")
    .onFallback()
        .to("file:./activemq-fallback-cache?fileName=msg-${date:now:yyyyMMddHHmmssSSS}.json");

熔断打开期间，消息直接存入缓存；熔断关闭（连接恢复）后，主路由恢复正常发送，同时缓存路由继续补发暂存的消息。

关键注意事项 #

• 本地缓存目录要保证读写权限，避免消息写入失败
• 如果需要比文件系统更高的可靠性，可将file组件替换为jdbc组件，把消息存入数据库，Camel会自动处理批量读取和顺序控制
• ActiveMQ端点开启usePersistentMessages=true，确保消息到达Broker后持久化，避免Broker重启丢失消息

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