ServiceActivators的Spring集成并发答案

【问题标题】：Spring Integration Concurrency of ServiceActivatorsServiceActivators的Spring集成并发
【发布时间】：2016-11-25 12:26:00
【问题描述】：

这是我之前的问题Spring Integration File reading 的延续。总之，我有一个 fileIn 通道（一个队列），然后是一个处理文件的 ServiceActivator 和一个用于保存文件的出站适配器。

我想引入并发，在多个线程中处理消息。我正在使用 java DSL（但不是 Java8）。我能够使用以下方式做到这一点......

@Bean
public MessageChannel fileInChannel() {
    return MessageChannels.queue("fileIn").get();
}

@Bean
public IntegrationFlow fileProcessingFlow() {
    return IntegrationFlows.from(fileInChannel())
            .handle(myFileProcessor, "processFile",
                    new Consumer<GenericEndpointSpec<ServiceActivatingHandler>>() {
                        @Override
                        public void accept(GenericEndpointSpec<ServiceActivatingHandler> t) {
                            t.poller(Pollers.fixedRate(100).maxMessagesPerPoll(1).taskExecutor(Executors.newCachedThreadPool()));
                        }
                    })
            .handle(Files.outboundAdapter(new File(outDir)).autoCreateDirectory(true).get())
            .get();
}

这行得通！我也试过了

public IntegrationFlow fileProcessingFlow() {
    return IntegrationFlows.from(fileInChannel())
            .channel(MessageChannels.executor(Executors.newCachedThreadPool()))
            .handle(myFileProcessor)
            .handle(Files.outboundAdapter(new File(outDir)).autoCreateDirectory(true).get())
            .get();
}

这也有效！我不知道这只是一种风格，还是一种方法比另一种更好。如果是这样，哪种方法更好。

其次，在上述情况下，“文件写入”（即最后一步）是顺序的还是会在不同的线程中工作。如果我也需要并发，我应该在句柄（fileProcessor）和句柄（outBoundAdapter）之间引入另一个taskExecutor通道吗？最终 outboundadapter 将是一个远程文件 S3 适配器，因此问题

【问题讨论】：

标签： java concurrency spring-integration

【解决方案1】：

这只是风格，虽然我更喜欢第二种。文件适配器是线程安全的。

一般来说，写入是并行进行的。

唯一的例外是如果您使用FileExistsMode.APPEND 进行写入，在这种情况下，在写入期间会持有一个锁，并且如果文件名哈希到与另一个文件相同的锁（有 256 个锁）正在编写，然后它将在第一个完成时运行。

【讨论】：