如何在spring集成中并行同步处理？

Question

在Spring集成中是否可以保持通道同步（发送消息后获得确认）但同时处理更多消息（并行处理）而无需使用线程创建自己的代码（即ExecutorService执行和提交工作人员）并等待它们？我想通过FTP上传文件，但同时上传更多文件，而无需在代码中创建自己的线程。我需要知道所有文件何时上传（这就是我希望它是同步的原因）。是否可以通过Spring集成配置，如果可以，如何？

Answer 1

在Spring中，通过使用异步任务处理，这是非常可能的。

首先创建一个将异步执行任务的服务。这里请注意performTask方法上的Async注释，它将被spring扫描并标记为异步执行。

import java.util.concurrent.Future;

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

@Service
public class AsyncTask {

    @Async
    public Future<Result> performTask(String someArgument) {
        // put the business logic here and collect the result below
        Result result = new Result(); // this is some custom bean holding your result
        return new AsyncResult<Result>(result);
    }
}

接下来创建一个组件（可选-也可以来自任何其他现有服务），该组件将调用上述服务。

import java.util.concurrent.Future;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class AsyncClass {

    @Autowired
    private AsyncTask asyncTask;

    public void doAsyncOperation() throws Exception {

    List<Future<Result>> futures = new ArrayList<Future<Result>>();

    for (int i = 1; i < 10; i++) {
        // Simulate multiple calls
        Future<Result > future = doAsync(String.valueOf(i));            
        futures.add(future);
    }

    for (Future<Result > future : futures) {
            // fetch the result
            Result result = future.get();
            // process the result
    }
}

    private Future<Result> doAsync(final String someArgument) {

        // this will immediately return with a placeholder Future object which
        // can be used later to fetch the result
        Future<Result> future = asyncTask.performAsync(someArgument);
        return future;
    }
}

启用异步所需的xml配置示例如下（对于基于注释的配置，请使用@EnableAsync）

<task:annotation-driven executor="myExecutor" />
<task:executor id="myExecutor" pool-size="30" rejection-policy="CALLER_RUNS"/>

有关详细文档，请参阅此处

Answer 2

看起来你需要一些流，比如：

>

<代码>

<代码>

<代码>

<代码>

<代码>

如果有什么不清楚，请告诉我。

更新

如何在Spring集成Java DSL中做到这一点？有什么例子吗？

像这样的东西：

@Configuration
@EnableIntegration
@IntegrationComponentScan
public class Configuration {

    @Bean
    public IntegrationFlow uploadFiles() {
        return f ->
                   f.split()
                       .handle(Ftp.outboundGateway(this.ftpSessionFactory,
                           AbstractRemoteFileOutboundGateway.Command.PUT, "'remoteDirectory'"))
                       .aggregate();
    }

}

@MessagingGateway(defaultRequestChannel = "uploadFiles.input") 
interface FtpUploadGateway {

    List<String> upload(List<File> filesToUpload);

}