Spring Integration：重复处理同一文件时FileAggregator未被调用

从你的代码和现象来看，核心问题出在聚合器的组生命周期管理或者拆分器标记的传递链路上，以下是具体的排查方向和解决办法：

1. 聚合器组未自动清理导致重复处理时被忽略 #

你使用的FileAggregator默认基于文件名（或相对路径）作为关联键（correlationKey）聚合消息组，且默认的消息存储（SimpleMessageStore）不会自动清理已完成的聚合组。如果第一次处理完文件后，对应的聚合组仍在缓存中，当相同文件名的文件再次被处理时，新的FileMarker.START标记进入聚合器后会被识别为已完成的组，从而被忽略，导致后续消息无法触发聚合逻辑。

解决办法：给聚合器配置组超时时间，让已完成的组自动过期清理：

.aggregate(new FileAggregator(), aggregatorSpec -> 
    aggregatorSpec.groupTimeout(60000) // 1分钟后自动清理过期组
)

2. 验证拆分器的标记与消息是否完整传递 #

已处理文件再次进入流程后，需要确认拆分器是否正常生成了完整的START、行内容、END标记，且这些标记都正确发送到了aggregatorChannel：

• 在aggregatorChannel前添加日志拦截，打印所有通过该通道的消息：

  .channel("aggregatorChannel")
  .handle(message -> {
      System.out.println("Received message for aggregation: " + message.getPayload());
      return message;
  })
  .aggregate(new FileAggregator())
  

  对比新文件和已处理文件的日志，排查是否缺少END标记，或者行内容是否被异常过滤。

3. 检查文件内容与过滤器逻辑 #

如果已处理文件的内容发生变化（比如所有行变为空字符串），第二个filter(StringUtils::isNotBlank)会将所有行内容丢弃到aggregatorChannel，但拆分器仍会发送END标记，聚合器应该依然会触发聚合（只是聚合结果为空）。若此时仍未触发聚合，说明END标记没有正常到达，需要排查拆分器的文件读取逻辑是否异常。

4. 确认Poller的重复拾取配置 #

虽然你提到Poller已拾取到文件，但仍需确认是否使用了AcceptOnceFileListFilter之类的幂等过滤器。如果文件的元数据（文件名、修改时间、大小）未变化，这类过滤器会阻止文件被重复拾取；若你需要强制重复处理，需调整过滤器配置为允许重复：

@Bean
public FileListFilter<File> fileListFilter() {
    return new AcceptOnceFileListFilter<>() {
        @Override
        public boolean accept(File file) {
            // 移除已处理记录，允许重复拾取
            this.remove(file);
            return super.accept(file);
        }
    };
}

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