Spring集成-来自sftp入站的聚合器

Question

从包含多个文件的 sftp 入站消息源聚合一条消息的最佳解决方案是什么？ 我们在远程机器上需要接收 3 个文件。之后，我们将这些文件的内容组合成一条 json 消息并转发。

public IntegrationFlow sftpIntegrationFlowBean() {
    final Map<String, Object> headers = new HashMap<>();
    headers.put("sftpFile", "sftpFile");
    final Consumer<AggregatorSpec> aggregator = t -> {
        t.sendPartialResultOnExpiry(true);
        t.expireGroupsUponCompletion(true);
        t.processor(new CustomMessageAggregator());
    };
    return IntegrationFlows
            .from(sftpInboundMessageSource(),
                    e -> e.id("sftpIntegrationFlow").poller(pollerMetadataSftp))
            .enrichHeaders(headers).aggregate(aggregator)
            .handle(customMessageSender).get();
}

轮询器每 15 分钟轮询一次。 在运行这段代码时，接下来会发生：

1. 检索文件并处理其中一个
2. 15 分钟后处理第二个文件
3. 再过 15 分钟处理第三个文件
4. 最后在 15 分钟后，消息被发送到目的地

如何在一个操作中完成所有这些操作而不会出现延迟？我确实使用 FileReadingMessageSource 尝试过，但结果相同。

提前谢谢你。

Answer 1

在PollerMetadata 中增加maxMessagesPerPoll。