Spring Integration：如何获取split()与aggregate()统计信息

一、调整日志级别获取基础运行数据 #

• 将org.springframework.integration下拆分器、聚合器相关包的日志级别设为DEBUG或TRACE：
  • DEBUG级别会输出拆分的总消息数、聚合进度（如已接收子消息数、待处理数量）等核心信息
  • TRACE级别会输出每条子消息的处理耗时、拆分/聚合的具体执行步骤，适合定位单条消息的延迟问题
  • 以Logback为例的配置示例：

    <logger name="org.springframework.integration.splitter" level="DEBUG"/>
    <logger name="org.springframework.integration.aggregator" level="DEBUG"/>
    

  • 注意：TRACE级别会生成大量日志，1万条消息场景下建议临时启用，或配合日志过滤规则（如只关注特定消息ID前缀）避免磁盘过载

二、利用内置Metrics采集标准化指标 #

Spring Integration 5.0+支持Micrometer集成，可直接采集split/aggregate的核心统计指标：

• 可采集的指标包括：拆分总消息数、聚合完成组数、聚合超时组数、子消息处理耗时分布等
• 配置步骤：
  1. 引入spring-boot-starter-actuator和micrometer-core依赖
  2. 配置Metrics工厂绑定注册中心：

     @Bean
     public IntegrationMetricsFactory integrationMetricsFactory(MeterRegistry meterRegistry) {
         return new IntegrationMetricsFactory(meterRegistry);
     }
     

  3. 通过Actuator端点/actuator/metrics查看指标，关键指标示例：
     • spring.integration.splitter.messages.split：累计拆分消息总数
     • spring.integration.aggregator.messages.completed：已完成聚合的消息组数
     • spring.integration.aggregator.messages.expired：超时未完成聚合的消息数

三、自定义拦截器统计关键维度数据 #

针对split和aggregate组件添加自定义拦截器，按需记录业务关注的统计项：

• 拆分拦截器示例（统计拆分数量与累计耗时）：

  public class SplitterStatsInterceptor extends ChannelInterceptorAdapter {
      private final AtomicInteger totalSplit = new AtomicInteger(0);
      private final AtomicLong totalCost = new AtomicLong(0);
  
      @Override
      public Message<?> preSend(Message<?> message, MessageChannel channel) {
          long start = System.currentTimeMillis();
          Message<?> result = super.preSend(message, channel);
          totalCost.addAndGet(System.currentTimeMillis() - start);
          int current = totalSplit.incrementAndGet();
          if (current % 1000 == 0) {
              System.out.printf("已拆分 %d 条消息，累计耗时 %d ms%n", current, totalCost.get());
          }
          return result;
      }
  }
  

• 聚合拦截器可类似实现，统计已接收子消息数、聚合完成数、单组聚合耗时等
• 将拦截器绑定到split/aggregate的输出通道：

  @Bean
  public MessageChannel splitOutputChannel() {
      DirectChannel channel = new DirectChannel();
      channel.addInterceptor(new SplitterStatsInterceptor());
      return channel;
  }
  

四、实时跟踪聚合进度 #

通过MessageGroupStore可直接获取聚合组的实时状态：

• 注入MessageGroupStore后，调用getMessageGroups()获取所有待聚合的组
• 遍历组信息，查看已接收子消息数、预期总数、创建时间等，定位进度缓慢的聚合组：

  @Autowired
  private MessageGroupStore messageGroupStore;
  
  public void checkAggregationStatus() {
      Collection<MessageGroup> groups = messageGroupStore.getMessageGroups();
      for (MessageGroup group : groups) {
          System.out.printf("聚合组ID: %d，已接收 %d/%d 条消息，创建时间: %s%n",
                  group.getGroupId(), group.getSize(), group.getExpectedSize(),
                  new Date(group.getTimestamp()));
      }
  }
  

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