Camel路由使用多线程将文件从本地文件夹移动到sftp文件夹

Question

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')

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

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"}

请让我知道可以为此做些什么。

Answer 1

（免责声明：不是专家，而是学问……)

我有一个非常类似的场景，使用一个线程来拾取文件，并使用一个SEDA路由来实际并行处理它们。

类似这样的事情：

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

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