骆驼路线使用多个线程将文件从本地文件夹移动到 sftp 文件夹答案

【问题标题】：Camel route move files from local folder to sftp folder using multiple threads骆驼路线使用多个线程将文件从本地文件夹移动到 sftp 文件夹
【发布时间】：2018-05-26 14:31:57
【问题描述】：

所以我将文件从本地文件夹移动到 sftp 位置。这些文件也可能达到 500,000 个。我有用于编写器进程的 seda，其中多个编写器正在编写文件，但移动文件已成为瓶颈。以下是我目前的骆驼路线：

from(props.splitFolderUri)
            .routeId('move-files')
            .log(LoggingLevel.INFO, 'file : ${exchange.getIn().headers[\'CamelFileName\']} :: moving this split file to sftp location ')
            .to(props.remoteFolderUri)
            .log(LoggingLevel.INFO, 'file : ${exchange.getIn().headers[\'CamelFileName\']} :: moved this split file to mailbox')

我想让这条路线多线程，以便多个线程继续移动文件。我尝试使用如下线程（10）

rom(props.splitFolderUri)
        .threads(10)
        .routeId('move-files')
        .log(LoggingLevel.INFO, 'file : ${exchange.getIn().headers[\'CamelFileName\']} :: moving this split file to sftp location ')
        .to(props.remoteFolderUri)
        .log(LoggingLevel.INFO, 'file : ${exchange.getIn().headers[\'CamelFileName\']} :: moved this split file to mailbox')

但是，我重命名文件操作失败。经过一些研究，我发现这是发生的情况：这是场景：

Thread#1 retrieves file1, moves to process folder
Thread#2 retrieves same file: file1 at the same time. file1 is deleted
Thread#2 cannot find file1 in source directory, rename fails.
Thread#1 fails due to deleted file by Thread#2

因此，我尝试了 noop=true 并且它有效，但对我来说不是一个好的选择，因为容器中的文件应该保留，或者容器内的内存非常有限。那么，关于同样的问题，什么是好的解决方案。我是骆驼的新手，所以无法准确指出实现这一目标的最佳方法。

更新：我什至尝试了以下方式：

from(props.splitFolderUri)
   .to("seda:processSplitFiles")

from("seda:processSplitFiles?concurrentConsumers=10")
    .to(log...)
    .to(props.remoteFolderUri)
    ...

但我不断收到以下错误：

    {"@timestamp":"2018-05-27T22:25:48.220-05:00","@version":1,"message":"Error during commit. Exchange[ID-8c85902bd64c-60121-1527477152921-0-90793]. Caused by:w [org.apache.camel.component.file.GenericFileOperationFailedException - Cannot rename file: GenericFile[/apps/splitFiles/spl/stressTesting.txt_3843_sc_3843_t_10000] to: GenericFile[/apps/splitFiles/spl/.camel/stressTesting.txt_3843_sc_3843_t_10000]]","logger_name":"org.apache.camel.component.file.GenericFileOnCompletion","thread_name":"Camel (camel-1) thread #58 - seda://moveSplitFiles","level":"WARN","level_value":30000,"stack_trace":"org.apache.camel.component.file.GenericFileOperationFailedException: Cannot rename file: GenericFile[/apps/splitFiles/spl/stressTesting.txt_3843_sc_3843_t_10000] to: GenericFile[/apps/splitFiles/spl/.camel/stressTesting.txt_3843_sc_3843_t_10000]
 org.apache.camel.component.file.strategy.GenericFileProcessStrategySupport.renameFile(GenericFileProcessStrategySupport.java:115)
  org.apache.camel.component.file.strategy.GenericFileRenameProcessStrategy.commit(GenericFileRenameProcessStrategy.java:88)
   org.apache.camel.component.file.GenericFileOnCompletion.processStrategyCommit(GenericFileOnCompletion.java:127)
    org.apache.camel.component.file.GenericFileOnCompletion.onCompletion(GenericFileOnCompletion.java:83) org.apache.camel.component.file.GenericFileOnCompletion.onComplete(GenericFileOnCompletion.java:57)
     org.apache.camel.util.UnitOfWorkHelper.doneSynchronizations(UnitOfWorkHelper.java:104)
      org.apache.camel.impl.DefaultUnitOfWork.done(DefaultUnitOfWork.java:229)
       org.apache.camel.util.UnitOfWorkHelper.doneUow(UnitOfWorkHelper.java:65)
        org.apache.camel.processor.CamelInternalProcessor$UnitOfWorkProcessorAdvice.after(CamelInternalProcessor.java:654)
         org.apache.camel.processor.CamelInternalProcessor$UnitOfWorkProcessorAdvice.after(CamelInternalProcessor.java:609)
          org.apache.camel.processor.CamelInternalProcessor$InternalCallback.done(CamelInternalProcessor.java:239)
           org.apache.camel.processor.Pipeline.process(Pipeline.java:109)
            org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
             org.apache.camel.component.seda.SedaConsumer.sendToConsumers(SedaConsumer.java:298)
              org.apache.camel.component.seda.SedaConsumer.doRun(SedaConsumer.java:207)
               org.apache.camel.component.seda.SedaConsumer.run(SedaConsumer.java:154)
                java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
                 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
                  java.lang.Thread.run(Thread.java:748)
                  ","HOSTNAME":"8c85902bd64c"}

请告诉我可以为此做些什么。

【问题讨论】：

标签： java multithreading apache-camel

【解决方案1】：

（免责声明：不是专家，而是学习......）

我有一个非常相似的场景，它使用单个线程来提取文件并使用 seda-route 实际并行处理它们。

类似这样的：

from(props.splitFolderUri)
    .to("seda:processSplitFiles");

from("seda:processSplitFiles?concurrentConsumers=10")
    .to(log...)
    .to(props.remoteFolderUri)
    ...

【讨论】：

【解决方案2】：

您可以使用骆驼聚合器以线程安全的方式工作。请在下面找到聚合器的链接：

http://camel.apache.org/aggregator.html

【讨论】：