为每个 Ftp 文件同时运行 Spring Integration 流程

Question

我有一个使用 Java DSL 配置的集成流，它使用 Ftp.inboundChannelAdapter 从 Ftp 服务器提取文件，然后将其转换为 JobRequest，然后我有一个触发我的批处理作业的 .handle() 方法，一切都按要求工作但是对于 FTP 文件夹中的每个文件顺序运行的过程

我在我的 Transformer Endpoint 中添加了 currentThreadName，它为每个文件打印相同的线程名称

这是我到目前为止所尝试的

1.task执行器bean

 @Bean
    public TaskExecutor taskExecutor(){
        return new SimpleAsyncTaskExecutor("Integration");

    }

2.集成流程

  @Bean
public IntegrationFlow integrationFlow(JobLaunchingGateway jobLaunchingGateway) throws IOException {
    return IntegrationFlows.from(Ftp.inboundAdapter(myFtpSessionFactory)
                    .remoteDirectory("/bar")
                    .localDirectory(localDir.getFile())
            ,c -> c.poller(Pollers.fixedRate(1000).taskExecutor(taskExecutor()).maxMessagesPerPoll(20)))
            .transform(fileMessageToJobRequest(importUserJob(step1())))
            .handle(jobLaunchingGateway)
            .log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload")
            .route(JobExecution.class,j->j.getStatus().isUnsuccessful()?"jobFailedChannel":"jobSuccessfulChannel")
            .get();
}

3.我还在另一个 SO 线程中读到我需要 ExecutorChannel，所以我配置了一个，但我不知道如何将此通道注入我的 Ftp.inboundAdapter，从日志中看到该通道始终为 @987654329 @ 我猜是DirectChannel

 @Bean
public MessageChannel inputChannel() {
    return new ExecutorChannel(taskExecutor());
}

我不知道我在这里缺少什么，或者我可能没有正确理解 Spring 消息系统，因为我对 Spring 和 Spring-Integration 非常陌生

感谢任何帮助

谢谢

Answer 1

您可以简单地将ExecutorChannel 注入到流程中，它将被框架应用到SourcePollingChannelAdapter。因此，将 inputChannel 定义为 bean，您只需这样做：

.channel(inputChannel())

在您的 .transform(fileMessageToJobRequest(importUserJob(step1()))) 之前。 在文档中查看更多信息：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl-channels

另一方面，要根据您的.taskExecutor(taskExecutor()) 配置并行处理您的文件，您只需要将.maxMessagesPerPoll(20) 设置为1。 AbstractPollingEndpoint中的逻辑是这样的：

this.taskExecutor.execute(() -> {
                int count = 0;
                while (this.initialized && (this.maxMessagesPerPoll <= 0 || count < this.maxMessagesPerPoll)) {
                    if (pollForMessage() == null) {
                        break;
                    }
                    count++;
                }

所以，我们确实有并行任务，但只有当它们到达 maxMessagesPerPoll 时，在您当前的情况下它是 20。文档中也有一些解释：https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#endpoint-pollingconsumer

maxMessagesPerPoll 属性指定在给定轮询操作中接收的最大消息数。这意味着轮询器继续调用 receive() 而不等待，直到返回 null 或达到最大值。例如，如果轮询器有一个 10 秒间隔触发器和 25 的 maxMessagesPerPoll 设置，并且它正在轮询一个在其队列中有 100 条消息的通道，则可以在 40 秒内检索所有 100 条消息。它抓取 25，等待 10 秒，然后抓取下一个 25，依此类推。