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对象。

合并未来对象时，使用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

服务程序

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

服务程序

@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));
    }
}

创建一个示例控制器进行测试

结果控制器.java

@RestController
public class ResultController {

    @Autowired
    private ResultService resultService;

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

}