Spring Integration：如何配置MessageSource避免重复读取并灵活指定输入？

近几周我在学习Spring Integration，发现官方文档里没提到各类MessageSource如何配置防止重复读取输入的机制。

以FTP服务器场景为例，我的集成流代码如下：

return IntegrationFlows.from("ftp-endpoint")
    .handle(fileProcessor())
    .get();

初始FTP目录结构：

readingDir/
    file1.txt
    file2.txt
    file3.txt

定时轮询的FTP MessageSource（ftp-endpoint）会读取这3个文件。之后目录新增文件，结构变为：

readingDir/
    file1.txt
    file2.txt
    file3.txt
    file4.txt
    file5.txt

核心疑问 #

• 如何配置才能让下次轮询时跳过已读取的file1.txt、file2.txt、file3.txt，只读取新增的file4.txt、file5.txt？
• 如果需要重新读取file3.txt，同时读取新文件但跳过file1.txt、file2.txt，该怎么配置？
• 希望这个策略能统一应用在所有轮询端点（邮件、DropBox、S3等），用一致的API实现。

现有尝试的困惑 #

目前我只找到通过SourcePollingChannelAdapterSpec为端点添加Advice的方式：

IntegrationFlows.from("ftp-endpoint", sourcePollingChannelAdapterSpec)
    .handle(fileProcessor())
    .get();

然后调用setAdvice(Arrays.asList(specialAdvice))实现去重逻辑，但这种方式太繁琐，而且我并不喜欢Advice这类标记接口，想知道有没有更简洁的统一方案。

统一解决方案：使用MetadataStore跟踪已读取记录 #

Spring Integration提供了**MetadataStore**抽象作为统一的状态跟踪机制，所有轮询型MessageSource都支持通过它来记录已处理的输入，避免重复读取，这是官方推荐的通用方案，不需要自定义Advice。

1. 基础配置：全局防重复 #

首先，配置一个MetadataStore实现（开发阶段可用内存型SimpleMetadataStore，生产环境推荐用Redis、JDBC等持久化实现），然后为MessageSource绑定MetadataStore和唯一前缀：

以FTP为例，配置FTP MessageSource：

@Bean
public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
    FtpInboundFileSynchronizer synchronizer = new FtpInboundFileSynchronizer(ftpSessionFactory());
    synchronizer.setDeleteRemoteFiles(false); // 保留远程文件
    synchronizer.setRemoteDirectory("/readingDir");
    return synchronizer;
}

@Bean
@InboundChannelAdapter(channel = "ftpChannel", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> ftpMessageSource(MetadataStore metadataStore) {
    FtpInboundFileSynchronizingMessageSource source = 
        new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
    source.setLocalDirectory(new File("./local-reading-dir"));
    source.setAutoCreateLocalDirectory(true);
    // 绑定MetadataStore，用于跟踪已处理文件
    source.setMetadataStore(metadataStore);
    // 设置唯一前缀，避免不同端点的记录冲突
    source.setMetadataStorePrefix("ftp-reading-dir-");
    return source;
}

配置完成后，每次轮询时MessageSource会自动检查MetadataStore中的记录，只读取未被标记为已处理的文件；处理完成后，会将文件的唯一标识（比如文件名+修改时间）存入MetadataStore，下次轮询自动跳过已处理文件。

2. 实现指定文件重新读取 #

如果需要重新读取某个文件（比如file3.txt），只需从MetadataStore中删除对应的记录即可：

@Autowired
private MetadataStore metadataStore;

public void reprocessFile(String fileName) {
    String key = "ftp-reading-dir-" + fileName; // 对应之前设置的前缀+文件名
    metadataStore.remove(key);
}

调用该方法后，下次轮询时file3.txt会被重新读取，而file1.txt、file2.txt因为记录仍在MetadataStore中会被跳过，同时新的file4.txt、file5.txt会被正常读取。

3. 跨端点统一策略 #

所有轮询型MessageSource（如邮件的ImapMailReceiver、S3的S3InboundFileSynchronizingMessageSource等）都支持setMetadataStore和setMetadataStorePrefix方法，只需为每个端点绑定同一个MetadataStore实例，并设置唯一前缀，即可实现统一的状态跟踪。

比如S3的配置示例：

@Bean
@InboundChannelAdapter(channel = "s3Channel", poller = @Poller(fixedDelay = "10000"))
public MessageSource<File> s3MessageSource(MetadataStore metadataStore) {
    S3InboundFileSynchronizingMessageSource source = 
        new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer());
    source.setLocalDirectory(new File("./local-s3-dir"));
    source.setMetadataStore(metadataStore);
    source.setMetadataStorePrefix("s3-bucket-");
    return source;
}

为什么不用Advice？ #

Advice一般用于自定义复杂拦截逻辑（如重试、事务控制），而MetadataStore是Spring Integration为轮询型数据源提供的原生、通用去重方案，无需自定义拦截逻辑，代码更简洁，且能跨端点统一复用。

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