为每个Ftp文件并发运行Spring集成流

Question

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

我在Transformerendpoint中添加了CurrentThreadName它为每个文件打印相同的线程名称

这是我到现在为止所尝试的

1.任务执行器bean

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

    }

  @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();
}

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

感谢任何帮助

谢谢

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秒的间隔触发器和maxMessagesPerPoll设置25，并且轮询队列中有100条消息的通道，则可以在40秒内检索所有100条消息。它抓取25，等待10秒，抓取下一个25，以此类推。