1. 简介
在项目开发时，经常会遇到从不同的接口服务拉取数据并将其汇总到响应中。在微服务中，这些数据源通常是外部 REST API。在本篇文章中，我们将使用 Java 的 CompletableFuture 高效地并行请求多个外部 REST API 中的数据。同时，会对整个请求过程中的异常处理、请求超时进行详细的介绍。
2. 为什么要并行调用？
假设我们需要更新对象中的多个字段，每个字段的值都来自外部 REST 调用。一种方法也是最简单的方式是依次调用每个 API 来更新每个字段。
但是，等待一个 REST 调用完成后再启动另一个会增加服务的整体响应时间。例如，如果我们调用两个应用程序接口，每个需要 5 秒钟，那么总时间至少要 10 秒钟，因为第二个调用需要等待第一个调用完成。
相反，我们可以并行调用所有 API，这样总时间就是最慢（耗时最长）的 REST 调用时间。例如，一个调用需要 7 秒，另一个需要 5 秒。在这种情况下，我们将等待 7 秒，因为我们已经并行处理了所有内容，必须等待所有结果完成。
因此，并行化是减少服务响应时间、提高服务可扩展性和改善用户体验的绝佳选择。
3. 实战案例
1 定义用于更新的目标 POJO
public class Purchase {
private String orderDescription ;
private String paymentDescription ;
private String buyerName ;
private String orderId ;
private String paymentId ;
private String userId ;
// getters and setters
}该采购类有三个需要更新的字段，每个字段都需要通过 ID 进行不同的 REST 调用来查询。
接下来，先创建一个类，定义一个 RestTemplate Bean 和一个用于 REST 调用的域 URL：
@Component
public class PurchaseRestCallsAsyncExecutor {
private static final String BASE_URL = "http://www.pack.com" ;
private final RestTemplate restTemplate ;
public PurchaseRestCallsAsyncExecutor(RestTemplate restTemplate) {
this.restTemplate = restTemplate ;
}
}接下来，分别编写3个REST接口调用的方法
现在，让我们来定义 /orders API 调用：
public String getOrderDescription(String orderId) {
ResponseEntity result = restTemplate.getForEntity(
String.format("%s/orders/%s", BASE_URL, orderId),
String.class) ;
return result.getBody() ;
} 然后，让我们定义 /payments API 调用：
public String getPaymentDescription(String paymentId) {
ResponseEntity result = restTemplate.getForEntity(
String.format("%s/payments/%s", BASE_URL, paymentId),
String.class) ;
return result.getBody() ;
} 最后，我们定义了 /users API 调用：
public String getUserName(String userId) {
ResponseEntity result = restTemplate.getForEntity(
String.format("%s/users/%s", BASE_URL, userId),
String.class) ;
return result.getBody() ;
} 这三个接口方法都使用 getForEntity() 方法进行 REST 调用，并将结果封装在一个 ResponseEntity 对象中。
3.2 使用 CompletableFuture 进行多次 REST 调用
现在，我们就可以创建一个方法，用于构建和运行一组三个 CompletableFutures：
public void updatePurchase(Purchase purchase) {
CompletableFuture.allOf(
CompletableFuture.supplyAsync(() -> getOrderDescription(purchase.getOrderId()))
.thenAccept(purchase::setOrderDescription),
CompletableFuture.supplyAsync(() -> getPaymentDescription(purchase.getPaymentId()))
.thenAccept(purchase::setPaymentDescription),
CompletableFuture.supplyAsync(() -> getUserName(purchase.getUserId()))
.thenAccept(purchase::setBuyerName)
).join() ;
}我们使用allOf()方法来构建CompletableFuture的步骤。每个参数都是一个并行任务，这些任务以另一个通过REST调用及其结果构建的CompletableFuture的形式存在。
我们首先使用supplyAsync()方法提供了一个Supplier，从这个Supplier中我们将检索数据。然后，我们使用thenAccept()来消费supplyAsync()的结果，并将其设置到Purchase类中相应的字段上。
在allOf()方法结束时，我们只是构建了这些任务，但尚未执行任何操作。
最后，我们在所有任务构建完毕后调用join()方法来并行运行所有任务并收集它们的结果。由于join()是一个线程阻塞操作，我们只在最后调用它，而不是在每个任务步骤之后调用，这是为了通过减少线程阻塞来优化应用程序性能。
由于我们没有为supplyAsync()方法提供一个自定义的ExecutorService，因此所有任务都在同一个Executor中运行。默认情况下，Java使用ForkJoinPool.commonPool()。
建议为supplyAsync()方法指定一个自定义的ExecutorService是一个好习惯，这样我们可以对线程池参数有更多的控制。
3.3 错误处理
在分布式系统中，服务不可用或网络故障是很常见的。这些故障可能发生在外部 REST API 中，而我们作为该 API 的客户端却并不知情。例如，如果应用程序宕机，这就导致发送的请求将永远无法完成。
因此，我们可以使用 handle() 方法单独处理每个 REST 调用异常：
public CompletableFuture handle(
BiFunction fn) ;该方法的参数是一个 BiFunction，其中包含作为参数的上一个任务的结果和异常。 接下来我们将 handle() 步骤添加到 CompletableFuture 的一个步骤中
public void updatePurchaseHandlingExceptions(Purchase purchase) {
CompletableFuture.allOf(
CompletableFuture.supplyAsync(() -> getPaymentDescription(purchase.getPaymentId()))
.thenAccept(purchase::setPaymentDescription)
.handle((result, exception) -> {
if (exception != null) {
// 异常处理
return null ;
}
return result ;
})
).join() ;
}在示例中，handle() 从 thenAccept() 调用的 setPaymentDescription() 中获取一个 Void 类型，然后将 thenAccept() 动作中抛出的任何错误存储到异常中。最后，如果没有异常抛出，则 handle() 返回作为参数传递的值。否则，返回空值。
3.4 处理 REST 调用超时
当我们使用 CompletableFuture 时，我们可以指定一个任务超时，类似于我们在 REST 调用中定义的超时。因此，如果任务没有在指定时间内完成，Java 会以超时异常（TimeoutException）结束任务执行，修改代码如下：
public void updatePurchaseHandlingExceptions(Purchase purchase) {
CompletableFuture.allOf(
CompletableFuture.supplyAsync(() -> getOrderDescription(purchase.getOrderId()))
.thenAccept(purchase::setOrderDescription)
// 设置超时时间5s
.orTimeout(5, TimeUnit.SECONDS)
.handle((result, exception) -> {
if (exception instanceof TimeoutException) {
// 异常处理
return null ;
}
return result ;
})
).join() ;
}我们在 CompletableFuture 中通过 orTimeout() 方法设置超时时间，如果在 5 秒内未完成任务时停止任务执行。同时还在 handle() 方法中添加了 if 语句，以便单独处理 TimeoutException。在 CompletableFuture 中添加超时可确保任务始终完成。这对于避免线程无限期地等待可能永远不会完成的操作结果非常重要。因此，它减少了处于长时间运行状态的线程数量，提高了应用程序的健康度。
