java - Spring Boot 异步与多线程

Question

我有一个 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 或多线程，但不确定如何在多线程中等待和运行 Async 中的两个进程并在这两个之后继续执行服务调用完毕!

任何建议或线索表示赞赏! TIA :)

Answer 1

你需要使用spring的AsyncResult类来包装你的结果，然后使用它的方法 .completable()返回 CompletableFuture目的。
合并 future 对象时使用 CompletableFuture.thenCompose()和 CompletableFuture.thenApply()合并数据的方法如下:

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

这是一个基本示例:
用 @EnableAsync 注释 Spring boot 主类注解

@SpringBootApplication
@EnableAsync
public class StackOverflowApplication {

    public static void main(String[] args) {
        SpringApplication.run(StackOverflowApplication.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
    }
}

创建另一个将合并其他两个服务数据的服务
结果服务.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));
    }
}

创建一个示例 Controller 进行测试
结果 Controller .java

@RestController
public class ResultController {

    @Autowired
    private ResultService resultService;

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

}