知識庫 / Reactive RSS 訂閱

1. 概述

通常在我們的應用程序中執行 HTTP 請求時，我們會按順序執行這些調用。然而，在某些情況下，我們可能希望同時執行這些請求。

例如，我們可能希望在從多個來源檢索數據或僅僅為了提高應用程序的性能時這樣做。

在本快速教程中，我們將探討幾種方法，以通過使用 Spring reactive WebClient 並行調用服務來完成此操作。

2. 反應式編程回顧

為了快速回顧，WebClient 在 Spring 5 中引入，並作為 Spring Web Reactive 模塊的一部分包含在其中。 它提供了一個反應式、非阻塞的 HTTP 請求發送接口。

要了解關於使用 WebFlux 進行反應式編程的深入指南，請查看我們優秀的 Spring 5 WebFlux 指南。

3. 一個簡單的用户服務

我們將在示例中利用一個簡單的 User API。 該 API 包含一個 GET 方法，用於通過將 ID 作為參數來檢索用户。

讓我們看看如何進行一次調用以根據給定的 ID 檢索用户：

WebClient webClient = WebClient.create("http://localhost:8080");

public Mono<User> getUser(int id) {
    LOG.info(String.format("Calling getUser(%d)", id));

    return webClient.get()
        .uri("/user/{id}", id)
        .retrieve()
        .bodyToMono(User.class);
}

在下一部分，我們將學習如何併發調用此方法。

4. 同時調用 WebClient 方法

在本節中，我們將看到幾個調用我們 getUser 方法的併發示例。我們還將研究 publisher 的兩種實現：Flux 和 Mono 在示例中的用法。

4.1. 同一服務多次調用

現在，我們假設我們想要同時獲取關於五名用户的詳細信息，並將結果以用户列表的形式返回：

public Flux fetchUsers(List userIds) {
    return Flux.fromIterable(userIds)
        .flatMap(this::getUser);
}

讓我們分解步驟，以瞭解我們已經做了什麼：

我們首先使用列表中的 userIds，通過靜態的 fromIterable 方法創建一個 Flux。

接下來，我們調用 flatMap 運行我們之前創建的 getUser 方法。這個反應式運算符的默認併發級別為 256，這意味着最多同時執行 256 個 getUser 調用。這個數字可以通過方法參數使用超載版本的 flatMap 進行配置。

值得注意的是，由於操作是在並行發生的，因此我們不知道結果的順序。如果我們需要保持輸入順序，可以使用 flatMapSequential 運算符代替。

由於 Spring WebClient 在底層使用非阻塞 HTTP 客户端，因此用户無需定義任何 Scheduler。 WebClient 負責調度調用並在其適當的線程上發佈結果，而無需阻塞。

4.2. 同時調用不同服務並返回相同類型的數據

現在我們來看如何同時調用多個服務。

在本示例中，我們將創建一個端點，該端點返回相同類型的 User。

public Mono<User> getOtherUser(int id) {
    return webClient.get()
        .uri("/otheruser/{id}", id)
        .retrieve()
        .bodyToMono(User.class);
}

現在，執行兩個或多個並行調用的方法變為：

public Flux fetchUserAndOtherUser(int id) {
    return Flux.merge(getUser(id), getOtherUser(id));
}

在這個示例中的主要區別在於，我們使用了靜態方法 merge 而不是 fromIterable 方法。 使用 merge 方法，我們可以將兩個或多個 Flux 合併為一個結果。

public Mono fetchUserAndItem(int userId, int itemId) {
    Mono user = getUser(userId);
    Mono item = getItem(itemId);

    return Mono.zip(user, item, UserWithItem::new);
}

@Test
public void givenClient_whenFetchingUsers_thenExecutionTimeIsLessThanDouble() {
        
    int requestsNumber = 5;
    int singleRequestTime = 1000;

    for (int i = 1; i <= requestsNumber; i++) {
        stubFor(get(urlEqualTo("/user/" + i)).willReturn(aResponse().withFixedDelay(singleRequestTime)
            .withStatus(200)
            .withHeader("Content-Type", "application/json")
            .withBody(String.format("{ \"id\": %d }", i))));
    }

    List<Integer> userIds = IntStream.rangeClosed(1, requestsNumber)
        .boxed()
        .collect(Collectors.toList());

    Client client = new Client("http://localhost:8089");

    long start = System.currentTimeMillis();
    List<User> users = client.fetchUsers(userIds).collectList().block();
    long end = System.currentTimeMillis();

    long totalExecutionTime = end - start;

    assertEquals("Unexpected number of users", requestsNumber, users.size());
    assertTrue("Execution time is too big", 2 * singleRequestTime > totalExecutionTime);
}