入站通道适配器：轮询多个服务器和目录

DeepSeek V3 中英对照 Inbound Channel Adapters: Polling Multiple Servers and Directories

从 5.0.7 版本开始，RotatingServerAdvice 功能可用；当配置为轮询器通知时，入站适配器可以轮询多个服务器和目录。配置该通知并将其添加到轮询器的通知链中，操作方式与常规通知相同。使用 DelegatingSessionFactory 来选择服务器，更多信息请参阅委托会话工厂。该通知的配置包含一个 RotationPolicy.KeyDirectory 对象列表。

@Bean
public RotatingServerAdvice advice() {
    List<RotationPolicy.KeyDirectory> keyDirectories = new ArrayList<>();
    keyDirectories.add(new RotationPolicy.KeyDirectory("one", "foo"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("one", "bar"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("two", "baz"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("two", "qux"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("three", "fiz"));
    keyDirectories.add(new RotationPolicy.KeyDirectory("three", "buz"));
    return new RotatingServerAdvice(delegatingSf(), keyDirectories);
}

此建议将轮询服务器 one 上的目录 foo，直到没有新文件存在，然后移动到服务器 two 上的目录 bar，接着是目录 baz，依此类推。

可以通过 fair 构造参数修改此默认行为：

@Bean
public RotatingServerAdvice advice() {
    ...
    return new RotatingServerAdvice(delegatingSf(), keyDirectories, true);
}

在这种情况下，无论之前的轮询是否返回了文件，建议都会移动到下一个服务器/目录。

或者，你也可以提供自定义的 RotationPolicy 来按需重新配置消息源：

public interface RotationPolicy {

    void beforeReceive(MessageSource<?> source);

    void afterReceive(boolean messageReceived, MessageSource<?> source);

}

和

@Bean
public RotatingServerAdvice advice() {
    return new RotatingServerAdvice(myRotationPolicy());
}

synchronizer 上的 local-filename-generator-expression 属性（localFilenameGeneratorExpression）现在可以包含 #remoteDirectory 变量。这允许从不同目录检索的文件下载到本地的类似目录中：

@Bean
public IntegrationFlow flow() {
    return IntegrationFlow.from(Sftp.inboundAdapter(sf())
                    .filter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
                    .localDirectory(new File(tmpDir))
                    .localFilenameExpression("#remoteDirectory + T(java.io.File).separator + #root")
                    .remoteDirectory("."),
                e -> e.poller(Pollers.fixedDelay(1).advice(advice())))
            .channel(MessageChannels.queue("files"))
            .get();
}

important

使用此建议时，不要在轮询器上配置 TaskExecutor；更多信息请参阅消息源的条件轮询器。

另请参阅便捷的 AbstractRemoteFileStreamingMessageSource.clearFetchedCache() API，当并非所有已获取的文件在单个轮询周期内处理完毕，但 SessionFactory 可能已轮换为不同实例时。