使用 OpenFeign 和 CompletableFuture 并行处理多个 HTTP 请求

教程

1、简介

在处理分布式系统时，调用外部服务并保持低延迟是一项至关重要的任务。

本文将带你了解如何使用 OpenFeign 和 CompletableFuture 来并行处理多个 HTTP 请求，处理错误，并设置网络和线程超时。

2、示例项目

为了说明并行请求的用法，我们要实现一个功能，允许客户在网站上购买物品。首先，该服务发出一个请求，根据客户所在国家获取可用的付款方式。其次，它发送一个请求给客户生成有关购买的报告。购买报告不包括有关付款方式的信息。

先添加 spring-cloud-starter-openfeign 依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>

3、创建调用外部依赖的客户端

使用 @FeignClient 注解创建两个指向 localhost:8083 的客户端：

@FeignClient(name = "paymentMethodClient", url = "http://localhost:8083")
public interface PaymentMethodClient {

    @RequestMapping(method = RequestMethod.GET, value = "/payment_methods")
    String getAvailablePaymentMethods(@RequestParam(name = "site_id") String siteId);
}

第一个客户端名为 paymentMethodClient。它调用 GET /payment_methods，使用代表客户所在国家/地区的 site_id 请求参数获取可用的付款方式。

第二个客户端如下：

@FeignClient(name = "reportClient", url = "http://localhost:8083")
public interface ReportClient {

    @RequestMapping(method = RequestMethod.POST, value = "/reports")
    void sendReport(@RequestBody String reportRequest);
}

我们将其命名为 reportClient，它调用 POST /reports 生成购买报告。

4、创建并行请求 Executor

根据需求，我们要依次调用两个客户端，在这种情况下，该 API 的总响应时间至少是两个请求的响应时间之和。

值得注意的是，报告中不包含付款方式的信息，因此这两个请求是独立的。因此，我们可以将任务并行化，将 API 的总响应时间缩短到与最慢请求的响应时间大致相同。

4.1、创建并行 Executor

使用 CompletableFutures 创建将两个请求并行化的服务：

@Service
public class PurchaseService {

    private final PaymentMethodClient paymentMethodClient;
    private final ReportClient reportClient;

    // 全参构造函数

    public String executePurchase(String siteId) throws ExecutionException, InterruptedException {
        CompletableFuture<String> paymentMethodsFuture = CompletableFuture.supplyAsync(() -> 
          paymentMethodClient.getAvailablePaymentMethods(siteId));
        CompletableFuture.runAsync(() -> reportClient.sendReport("Purchase Order Report"));

        return String.format("Purchase executed with payment method %s", paymentMethodsFuture.get());
    }
}

executePurchase() 方法首先发布一个并行任务，使用 supplyAsync() 获取可用的付款方式。然后，提交另一个并行任务，使用 runAsync() 生成报告。最后，使用 get() 获取支付方法结果并返回完整结果。

之所以选择 supplyAsync() 和 runAsync() 这两个任务，是因为这两个方法的性质不同。supplyAsync() 方法会返回 GET 调用的结果。而，runAsync() 不会返回任何结果，因此更适合用于生成报告。

另一个区别是，runAsync() 在调用代码时会立即启动一个新线程，而无需线程池进行任何任务调度。相比之下，supplyAsync() 任务可能会被调度或延迟，这取决于线程池是否调度了其他任务。

使用 WireMock 进行集成测试：

@BeforeEach
public void startWireMockServer() {
    wireMockServer = new WireMockServer(8083);
    configureFor("localhost", 8083);
    wireMockServer.start();

    stubFor(post(urlEqualTo("/reports"))
      .willReturn(aResponse().withStatus(HttpStatus.OK.value())));
}

@AfterEach
public void stopWireMockServer() {
    wireMockServer.stop();
}

@Test
void givenRestCalls_whenBothReturnsOk_thenReturnCorrectResult() throws ExecutionException, InterruptedException {
    stubFor(get(urlEqualTo("/payment_methods?site_id=BR"))
      .willReturn(aResponse().withStatus(HttpStatus.OK.value()).withBody("credit_card")));

    String result = purchaseService.executePurchase("BR");

    assertNotNull(result);
    assertEquals("Purchase executed with payment method credit_card", result);
}

在上面的测试中，我们首先使用 @BeforeEach 和 @AfterEach 注解配置了一个 WireMockServer，使其在 localhost:8083 启动，并在完成后关闭。

然后，在测试方法中，我们使用了两个 stub，当我们调用这两个 feign 客户端时，它们都会响应 200 HTTP 状态。最后，我们使用 assertEquals() 来断言并行 executor 的结果是正确的。

4.2、使用 exceptionally() 处理外部错误

如果 GET /payment_methods 请求以 404 HTTP 状态失败，表明该国家没有可用的支付方法，该怎么办？在这种情况下需要做一些处理，例如，返回一个默认值。

要在 CompletableFuture 中处理错误，需要在 paymentMethodsFuture 中添加以下 exceptionally() 代码块：

CompletableFuture <String> paymentMethodsFuture = CompletableFuture.supplyAsync(() -> paymentMethodClient.getAvailablePaymentMethods(siteId))
  .exceptionally(ex -> {
      if (ex.getCause() instanceof FeignException && 
             ((FeignException) ex.getCause()).status() == 404) {
          return "cash";
      });

现在，如果得到 404，就会返回名为 cash（现金）的默认付款方式：

@Test
void givenRestCalls_whenPurchaseReturns404_thenReturnDefault() throws ExecutionException, InterruptedException {
    stubFor(get(urlEqualTo("/payment_methods?site_id=BR"))
        .willReturn(aResponse().withStatus(HttpStatus.NOT_FOUND.value())));

    String result = purchaseService.executePurchase("BR");

    assertNotNull(result);
    assertEquals("Purchase executed with payment method cash", result);
}

5、为并行任务和网络请求添加超时

在调用外部依赖项时，我们无法确定请求的运行时间。因此，如果请求耗时过长，就应该放弃该请求。考虑到这一点，我们可以添加两种类型：FeignClient 和 CompletableFuture 超时。

5.1、为 Feign 客户端添加网络超时

这种类型的超时适用于单个网络请求。

我们可以使用 Spring Boot 自动配置为 FeignClient 配置超时：

feign.client.config.paymentMethodClient.readTimeout: 200
feign.client.config.paymentMethodClient.connectTimeout: 100

在上述 application.properties 文件中，我们为 PaymentMethodClient 设置了读取和连接超时。数值以毫秒为单位。

连接超时会告诉 feign 客户端在达到阈值后中断 TCP 握手连接尝试。同样，当连接正常建立，但协议无法从 Socket 读取数据时，读取超时会中断请求。

然后，我们就可以在并行 executor 中的 exceptionally() 块内处理这类错误：

if (ex.getCause() instanceof RetryableException) {
    // 处理 TCP 超时
    throw new RuntimeException("TCP call network timeout!");
}

添加另一个测试场景，验证行为的正确性：

@Test
void givenRestCalls_whenPurchaseRequestWebTimeout_thenReturnDefault() {
    stubFor(get(urlEqualTo("/payment_methods?site_id=BR"))
      .willReturn(aResponse().withFixedDelay(250)));

    Throwable error = assertThrows(ExecutionException.class, () -> purchaseService.executePurchase("BR"));

    assertEquals("java.lang.RuntimeException: REST call network timeout!", error.getMessage());
}

如上，我们使用 withFixedDelay() 方法，用 250 毫秒来模拟 TCP 超时。

5.2、添加线程超时

线程超时会停止整个 CompletableFuture，而不仅仅是一次请求尝试。例如，对于 feign 户端重试，在评估超时阈值时，原始请求和重试尝试的时间也会计算在内。

要配置线程超时，我们可以稍微修改一下我们的支付方法 CompletableFuture：

CompletableFuture<String> paymentMethodsFuture = CompletableFuture.supplyAsync(() -> paymentMethodClient.getAvailablePaymentMethods(siteId))
  .orTimeout(400, TimeUnit.MILLISECONDS)
  .exceptionally(ex -> {
       // 异常处理
   });

然后，我们就可以在 exceptionally() 代码块中处理线程超时错误：

if (ex instanceof TimeoutException) {
    // 处理线程超时异常
    throw new RuntimeException("Thread timeout!", ex);
}

测试：

@Test
void givenRestCalls_whenPurchaseCompletableFutureTimeout_thenThrowNewException() {
    stubFor(get(urlEqualTo("/payment_methods?site_id=BR"))
        .willReturn(aResponse().withFixedDelay(450)));

    Throwable error = assertThrows(ExecutionException.class, () -> purchaseService.executePurchase("BR"));

    assertEquals("java.lang.RuntimeException: Thread timeout!", error.getMessage());
}

我们为 /payments_method 添加了更长的延迟，这样它就能通过网络超时阈值，但在线程超时时会失败。

6、总结

本文介绍了如何使用 CompletableFuture 和 FeignClient 并行执行多个外部请求，还介绍了如何设置网络超时和线程超时，以及如何优雅地处理超时异常。

Ref：https://www.baeldung.com/feign-client-completablefuture-spring-boot