使用spring入站文件适配器进行并发处理

Question

我有一个文件系统目录，我想轮询文件，然后并发处理每个文件，每个文件只有一个线程。我的印象是，在封面下，In边界文件适配器将每个文件放入队列中，这样我就可以使用下游的执行器通道来并发处理以后的调用。我在JavaConfig中实现如下：

return IntegrationFlows
            .from(s -> s.file(inboundMessageDirectory.toFile(), Comparator.comparing(File::lastModified)) // serve oldest first
                            .scanner(directoryScanner) // we know the directory structure, so we can take advantage of that with a custom scanner
                            .filter(new AcceptOnceFileListFilter<>(MAX_FILTER_CAPACITY)), // limit number of references in memory
                    e -> e.poller(Pollers
                            .fixedDelay(fileSystemPollDelay)
                            .get()))
            .channel(MessageChannels.executor(executor).get())
            .transform(File::toPath)
            .enrichHeaders(cleanUpConfigurer)
            .get()

执行器通道下游的每个通道本身就是一个直接通道。

然而，我发现下游服务的并发性很差。对于缓存的线程池，我看到相同的线程基本上是串行地执行下游代码，而如果使用固定的池执行器，我看到不同的线程在静态串行执行中进行权衡。

我还尝试在民意调查者和执行者频道之间架起一座桥梁，但没有成功。

Answer 1

这只是因为罩下的SourcePollingChannelAdapterFactoryBean：

if (this.pollerMetadata.getMaxMessagesPerPoll() == Integer.MIN_VALUE){
    // the default is 1 since a source might return
    // a non-null and non-interruptible value every time it is invoked
    this.pollerMetadata.setMaxMessagesPerPoll(1);
}

所以，每个都是您的。fixedDelay（fileSystemPollDelay）队列中只轮询一个文件进行处理。

所以，只需增加。maxMessagesPerPoll（）到适合您的系统值，并享受并发的乐趣！

顺便说一句，没有理由在轮询适配器之后引入ExecutorChannel。您只需为. poller（）使用. taskExecitor（）完全出于相同的并发原因。