Нужно обрабатывать несколько файлов параллельно в Spring Integration

Question

У меня есть каталог SFTP и чтение файлов и отправка файлов для дальнейшей обработки в ServiceActivator. В любой момент мне нужно обрабатывать их параллельно с помощью обработчика.

Вот моя интеграция с SPG java DSL.

IntegrationFlows.from(Sftp.inboundAdapter(getSftpSessionFactory()) 
         .temporaryFileSuffix("COPY") 
         .localDirectory(directory) 
         .deleteRemoteFiles(false) 
         .preserveTimestamp(true) 
         .remoteDirectory("remoteDir")) 
         .patternFilter("*.txt")), e -> e.poller(Pollers.fixedDelay(500).maxMessagesPerPoll(5))) 
         .handle("mybean", "myMethod") 
         .handle(Files.outboundAdapter(new File("success")))   
         .deleteSourceFiles(true) 
         .autoCreateDirectory(true)) 
         .get(); 

Update: Вот мой ThreadPoolExecutor:

@Bean(name = "executor") 
public Executor getExecutor() 
{ 
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
    executor.setCorePoolSize(4); 
    executor.setMaxPoolSize(4); 
    executor.setQueueCapacity(20);   
    executor.initialize(); 
    return executor; 
} 

Answer 1

Sftp.inboundAdapter() (SftpInboundFileSynchronizingMessageSource) возвращает удаленные файлы по одному в любом случае. Прежде всего, он синхронизирует их с локальным каталогом и только после этого опросит их для обработки сообщений как полезную нагрузку File.

Чтобы обрабатывать их параллельно, этого было бы достаточно, чтобы добавить taskExecutor в ваше определение e.poller(), и все эти maxMessagesPerPoll(5) будут распространяться по различным темам.