使用 WebClient 執行同步請求

1. 引言

本教程將指導您學習如何使用 WebClient 執行同步請求。

雖然響應式編程正日益普及，但我們將探討在哪些場景下，這些阻塞請求仍然適用且必要。

2. HTTP 客户端庫在 Spring 中的概述

首先，讓我們簡要回顧一下當前可用的客户端庫，並瞭解我們的 WebClient 的位置。

在 Spring Framework 3.0 中引入後，RestTemplate 因其簡單的模板方法 API 用於 HTTP 請求而流行。然而，其同步特性和大量的重載方法導致高流量應用程序中出現複雜性和性能瓶頸。

在 Spring 5.0 中，WebClient 作為更高效、反應式的替代方案，用於非阻塞請求而引入。雖然它屬於反應式 Web 框架的一部分，但它支持用於同步和異步通信的流暢 API。

在 Spring Framework 6.1 中，RestClient 提供了另一種執行 REST 調用選項。它結合了 WebClient 的流暢 API 以及 RestTemplate 的基礎設施，包括消息轉換器、請求工廠和攔截器。

RestClient 經過優化以執行同步請求，而 WebClient 在我們的應用程序也需要異步或流式功能時更合適。使用 WebClient 進行阻塞和非阻塞 API 調用，有助於我們保持代碼庫的一致性，並避免混合使用不同的客户端庫。

3. 阻塞式與非阻塞式 API 調用

在討論各種 HTTP 客户端時，我們使用了諸如同步和異步、阻塞式和非阻塞式等術語。這些術語具有上下文敏感性，有時可能代表同一個概念的不同名稱。

在方法調用上下文中，WebClient 支持基於其發送和接收 HTTP 請求和響應的方式，進行同步和異步交互。如果它在執行前一個請求之前等待其完成，然後再執行後續請求，則它是在阻塞式地執行，並且結果將同步返回。

另一方面，我們可以通過執行非阻塞調用來實現異步交互，該調用會立即返回。在等待另一個系統響應的同時，其他處理可以繼續進行，並且一旦準備好，結果將異步提供。

4. 使用同步請求的場景

如前所述，WebClient 是 Spring Webflux 框架的一部分，該框架默認採用響應式編程。但是，該庫提供了異步和同步操作的支持，使其適用於響應式和 Servlet 棧 Web 應用程序。

使用阻塞方式的 WebClient 在需要即時反饋的場景下是合適的，例如在測試或原型設計期間。這種方法使我們能夠專注於功能，然後再考慮性能優化。

許多現有應用程序仍然使用阻塞客户端，如 RestTemplate。由於 RestTemplate 從 Spring 5.0 版本開始進入維護模式，因此重構遺留代碼庫需要依賴項更新，並且可能需要過渡到非阻塞架構。在這種情況下，我們可以暫時使用 WebClient 以阻塞方式運行。

即使在新的項目中使用，某些應用程序部分也可以設計為同步流程。這可能包括向各種外部系統進行順序 API 調用，其中一個調用的結果是用於執行下一個調用的必要條件。 WebClient 可以處理阻塞和非阻塞調用，而不是使用不同的客户端。

如稍後所見，同步和異步執行之間的切換相對簡單。 在儘可能的情況下，我們應該避免使用阻塞調用，尤其是在我們正在工作於響應式堆棧時。

5. 使用 WebClient 進行同步 API 調用

當發送 HTTP 請求時，WebClient 將返回 Reactor Core 庫中的兩種反應式數據類型之一：Mono 或 Flux。這些返回類型代表數據流，其中 Mono 對應於單個值或空結果，而 Flux 則指零個或多個值的流。具有異步和非阻塞的 API 允許調用者決定何時以及如何訂閲，從而保持代碼的反應式。

`然而，如果我們想要模擬同步行為，我們可以調用可用的 block() 方法。 它會阻塞當前操作以獲取結果。

為了更準確地説，block() 方法會觸發對反應式流的新訂閲，從而從源頭到消費者啓動數據流。內部，它使用 CountDownLatch 等待流完成，該方法暫停當前線程，直到操作完成，即 Mono 或 Flux 發出結果。 block() 方法將非阻塞操作轉換為傳統的阻塞操作，從而導致調用線程等待結果。

6. 實際示例

讓我們看看它的實際應用。想象一下，一個 API 網關應用程序，位於客户端應用程序和兩個後端應用程序（客户和計費系統）之間。其中一個存儲客户信息，而另一個則提供計費詳情。不同的客户端通過北向 API 與我們的 API 網關交互，該接口向客户端公開，用於檢索客户信息，包括他們的計費詳情。

@GetMapping("/{id}")
CustomerInfo getCustomerInfo(@PathVariable("id") Long customerId) {
    return customerInfoService.getCustomerInfo(customerId);
}

以下是模型類的結構：

public class CustomerInfo {
    private Long customerId;
    private String customerName;
    private Double balance;

    // standard getters and setters
}

API 網關通過提供內部與客户和賬單應用程序的單個端點，簡化了流程。它還會從這兩個系統聚合數據。

考慮我們使用同步 API 在整個系統中的情況。然而，我們最近升級了客户和賬單系統，以處理異步和非阻塞操作。讓我們看看這兩個 southbound API 現在是什麼樣子的。

客户 API：

@GetMapping("/{id}")
Mono<Customer> getCustomer(@PathVariable("id") Long customerId) throws InterruptedException {
    TimeUnit.SECONDS.sleep(SLEEP_DURATION.getSeconds());
    return Mono.just(customerService.getBy(customerId));
}

計費 API：

@GetMapping("/{id}")
Mono<Billing> getBilling(@PathVariable("id") Long customerId) throws InterruptedException {
    TimeUnit.SECONDS.sleep(SLEEP_DURATION.getSeconds());
    return Mono.just(billingService.getBy(customerId));
}

在實際應用場景中，這些API將作為獨立的組件存在。為了簡化開發，我們已將它們組織成不同的軟件包。此外，為了測試，我們已引入延遲，以模擬網絡延遲：

public static final Duration SLEEP_DURATION = Duration.ofSeconds(2);

與兩個後端系統不同，我們的API網關應用程序必須暴露一個同步的、阻塞式的API，以避免破壞客户端合同。因此，那裏沒有任何變更。

業務邏輯位於 CustomerInfoService 中。首先，我們將使用 WebClient 從Customer系統檢索數據：

Customer customer = webClient.get()
  .uri(uriBuilder -> uriBuilder.path(CustomerController.PATH_CUSTOMER)
    .pathSegment(String.valueOf(customerId))
    .build())
  .retrieve()
  .onStatus(status -> status.is5xxServerError() || status.is4xxClientError(), response -> response.bodyToMono(String.class)
    .map(ApiGatewayException::new))
  .bodyToMono(Customer.class)
  .block();

接下來，請查看計費系統：

Billing billing = webClient.get()
  .uri(uriBuilder -> uriBuilder.path(BillingController.PATH_BILLING)
    .pathSegment(String.valueOf(customerId))
    .build())
  .retrieve()
  .onStatus(status -> status.is5xxServerError() || status.is4xxClientError(), response -> response.bodyToMono(String.class)
    .map(ApiGatewayException::new))
  .bodyToMono(Billing.class)
  .block();

最後，我們將會利用來自兩個組件的響應來構建一個響應：

new CustomerInfo(customer.getId(), customer.getName(), billing.getBalance());

如果其中一個API調用失敗，則在 onStatus() 方法中定義的錯誤處理機制會將HTTP錯誤狀態映射到 ApiGatewayException 異常。我們採用了一種傳統的做法，而不是通過 Mono.error() 方法的反應式替代方案。由於我們的客户端期望一個同步的API，因此我們拋出會傳播給調用者的異常。

儘管客户和計費系統具有異步特性，但 WebClient 的 block() 方法使我們能夠聚合來自這兩個來源的數據，並以透明的方式返回合併的結果給客户端。

6.1. 優化多個API調用

此外，由於我們同時向不同的系統發出兩個連續的調用，我們可以通過避免單獨阻塞每個響應來優化該過程。我們可以執行以下操作：

private CustomerInfo getCustomerInfoBlockCombined(Long customerId) {
    Mono<Customer> customerMono = webClient.get()
      .uri(uriBuilder -> uriBuilder.path(CustomerController.PATH_CUSTOMER)
        .pathSegment(String.valueOf(customerId))
        .build())
      .retrieve()
      .onStatus(status -> status.is5xxServerError() || status.is4xxClientError(), response -> response.bodyToMono(String.class)
        .map(ApiGatewayException::new))
      .bodyToMono(Customer.class);

    Mono<Billing> billingMono = webClient.get()
      .uri(uriBuilder -> uriBuilder.path(BillingController.PATH_BILLING)
        .pathSegment(String.valueOf(customerId))
        .build())
      .retrieve()
      .onStatus(status -> status.is5xxServerError() || status.is4xxClientError(), response -> response.bodyToMono(String.class)
        .map(ApiGatewayException::new))
      .bodyToMono(Billing.class);

    return Mono.zip(customerMono, billingMono, (customer, billing) -> new CustomerInfo(customer.getId(), customer.getName(), billing.getBalance()))
      .block();
}

zip() 是一個方法，它將多個 Mono 實例合併為一個單一的 Mono。新的 Mono 完成時，所有提供的 Mono 都已生成其值，然後根據指定函數聚合 – 在我們的案例中，創建 CustomerInfo 對象。這種方法更有效率，因為它允許我們同時等待來自兩個服務的合併結果。

為了驗證我們是否提高了性能，讓我們在兩個場景下運行測試：

@Autowired
private WebTestClient testClient;

@Test
void givenApiGatewayClient_whenBlockingCall_thenResponseReceivedWithinDefinedTimeout() {
    Long customerId = 10L;
    assertTimeout(Duration.ofSeconds(CustomerController.SLEEP_DURATION.getSeconds() + BillingController.SLEEP_DURATION.getSeconds()), () -> {
        testClient.get()
          .uri(uriBuilder -> uriBuilder.path(ApiGatewayController.PATH_CUSTOMER_INFO)
            .pathSegment(String.valueOf(customerId))
            .build())
          .exchange()
          .expectStatus()
          .isOk();
    });
}

最初，測試失敗了。但是，切換到等待合併結果後，測試在客户和計費系統調用的合併時長內完成。 這表明我們通過聚合來自兩個服務的響應來改進了性能。 即使我們採用阻塞同步方法，我們仍然可以遵循最佳實踐來優化性能。這有助於確保系統保持高效和可靠。

7. 結論

在本教程中，我們演示瞭如何使用 WebClient 管理同步通信，該工具專為響應式編程而設計，但也能發起阻塞調用。

總結一下，我們討論了在響應式堆棧中，使用 WebClient 相對於其他庫（如 RestClient）的優勢，特別是為了保持一致性並避免混合使用不同客户端庫。最後，我們還探討了通過聚合來自多個服務的響應，而不阻塞每個調用，從而優化性能。

知識庫 / Spring WebFlux RSS 訂閱