Spring Boot Async 与多线程

Posted 2023-02-26

【中文标题】Spring Boot Async 与多线程

【英文标题】：Spring boot Async with Multithreading

【发布时间】：2020-06-17 16:15:08

【问题描述】：

我有一个 Spring Boot 微服务，我们在其中调用多个服务（比如说服务 A 和服务 B）。我正在尝试根据某些条件在多个线程上异步调用这两个服务，一旦处理完成，我想合并来自服务 A 和服务 B 的响应。

我知道我们可以使用@Async 来异步运行一个进程，并使用 ExecutorService 为一个服务启动多个线程。

但我不确定如何将所有东西放在一起。所以在这里寻找任何建议？

              @Async
              Service A(thread1,thread2) \
MicroService /                             (Merge from Response of ServiceA and ServiceB)
             \ @Async
              Service B(thread1,thread2) /

我知道这主要是上面的理论解释，但我尝试关注/浏览多个网站，但大多数文章都解释了 Aync 或多线程，但不知道如何在多线程中等待和运行异步中的两个进程并继续执行这两个服务调用完成后！

感谢任何建议或线索！ TIA :)

【参考方案1】：

你需要使用spring的AsyncResult类来包装你的结果，然后使用它的方法.completable()返回CompletableFuture对象。

合并未来对象时，使用CompletableFuture.thenCompose() 和CompletableFuture.thenApply() 方法合并数据，如下所示：

CompletableFuture<Integer> result = futureData1.thenCompose(fd1Value -> 
                futureData2.thenApply(fd2Value -> 
                        merge(fd1Value, fd2Value)));

这是一个基本示例：

用@EnableAsync注解注解Spring boot主类

@SpringBootApplication
@EnableAsync
public class ***Application 

    public static void main(String[] args) 
        SpringApplication.run(***Application.class, args);
    

创建一个返回CompletableFuture的示例服务

Aservice.java

@Service
public class Aservice 

    @Async
    public CompletableFuture<Integer> getData() throws InterruptedException 
        Thread.sleep(3000); // sleep for 3 sec
        return new AsyncResult<Integer>(2).completable(); // wrap integer 2
    

Bservice.java

@Service
public class Bservice 

    @Async
    public CompletableFuture<Integer> getData() throws InterruptedException 
        Thread.sleep(2000); // sleep for 2 sec
        return new AsyncResult<Integer>(1).completable(); // wrap integer 1
    

创建另一个服务，它将合并其他两个服务数据

ResultService.java

@Service
public class ResultService 

    @Autowired
    private Aservice aservice;
    @Autowired
    private Bservice bservice;

    public CompletableFuture<Integer> mergeResult() throws InterruptedException, ExecutionException 
        CompletableFuture<Integer> futureData1 = aservice.getData();
        CompletableFuture<Integer> futureData2 = bservice.getData();

        // Merge futures from Aservice and Bservice
        return futureData1.thenCompose(
            fd1Value -> futureData2.thenApply(fd2Value -> fd1Value + fd2Value));
    

创建一个用于测试的示例控制器

ResultController.java

@RestController
public class ResultController 

    @Autowired
    private ResultService resultService;

    @GetMapping("/result")
    CompletableFuture<Integer> getResult() throws InterruptedException, ExecutionException 
        return resultService.mergeResult();
    

【讨论】：

为什么是Thread.join？为什么不让ResultController 只返回另一个带有聚合的CompletableFuture？

@EdwinDalorzo，我已根据您的评论更新了答案

@AjitSoman 和 Edwin 感谢您的回复。我会尝试了解这里发生的事情并尝试一下

@AjitSoman 再次感谢完整的示例。但看起来上面只是用于异步处理对吗？要打开多个线程，我可以使用 ExecutorService 吗？

@JingJong，是的，你需要像这样为Execotor创建bean：`@Bean public Executor taskExecutor() ThreadPoolTask​​Executor executor = new ThreadPoolTask​​Executor(); executor.setCorePoolSize(3); executor.setMaxPoolSize(15); executor.setQueueCapacity(500); executor.setThreadNamePrefix("GithubLookup-"); executor.initialize();返回执行者； `。查看此链接了解更多信息：spring.io/guides/gs/async-method

【参考方案2】：

你可以看看java的CompletableFuture。 ComplitableFuture 允许联合多个异步任务（也是 CompletableFuture）并等待所有 CompletableFutures 的结果。我不确定它是否完全适合您的情况，但它可能会有所帮助。 https://www.baeldung.com/java-completablefuture#Multiple