Блог

У меня есть поток интеграции, настроенный с использованием Java DSL, который извлекает файл с Ftp-сервера, используя Ftp.inboundChannelAdapter затем преобразует его в JobRequest , затем у меня есть .handle() метод, который запускает мое пакетное задание, все работает в соответствии с требованиями, но процесс выполняется последовательно для каждого файла внутри папки FTP

Я добавил currentThreadName в свою конечную точку Transformer, что он печатал одно и то же имя потока для каждого файла

Вот что я пробовал до сих пор

1. компонент-исполнитель задачи

  @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 , из журналов видно, что канал всегда integrationFlow.channel#0 , который, я думаю, является DirectChannel

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

Я не знаю, чего мне здесь не хватает, или я, возможно, неправильно понял систему обмена сообщениями Spring, поскольку я очень новичок в Spring и Spring-интеграции

Любая помощь приветствуется

Спасибо

ExecutorChannel Вы можете просто ввести в поток, и он будет применен к SourcePollingChannelAdapter фреймворку. Итак, inputChannel определив это как компонент, вы просто делаете это:

 .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) as 1 . Логика в этом AbstractPollingEndpoint такова:

 this.taskExecutor.execute(() -> {
                int count = 0;
                while (this.initialized amp;amp; (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 или не будет достигнуто максимальное значение. Например, если средство опроса имеет десятисекундный интервал срабатывания и значение maxMessagesPerPoll равное 25, и оно опрашивает канал, в очереди которого 100 сообщений, все 100 сообщений могут быть получены в течение 40 секунд. Он захватывает 25, ждет десять секунд, захватывает следующие 25 и так далее.