知識庫 / Spring WebFlux RSS 訂閱

然而，<em >WebClient</em> 還可以在 Servlet 3.1+ 容器（如 Jetty）上運行，但此時其工作方式有所不同。

如果我們比較在運行 WebFlux 應用程序時，使用 Jetty 以及不使用 <em >WebClient</em> 時創建的線程數量，我們會注意到一些額外的線程。

在這裏，<em >WebClient</em> 必須創建其自身的 <em >事件循環</em>。因此，我們可以看到這個事件循環創建的固定數量的處理線程：

在某些情況下，為客户端和服務器分別使用獨立的線程池可以提供更好的性能。雖然這並非 Netty 的默認行為，但如果需要，仍然可以為 <em >WebClient</em> 聲明一個專門的線程池。

我們將如何在後面的章節中看到這一點。

8. 數據訪問庫中的線程模型

正如我們之前所見，即使一個簡單的應用程序通常由多個需要連接的部分組成。

這些部分中的典型示例包括數據庫和消息代理。目前連接到許多這些庫仍然是阻塞式的，但這種情況正在迅速變化。

現在有幾個數據庫提供用於連接的反應式庫。許多這些庫在 Spring Data 中可用，我們也可以直接使用它們。

這些庫使用的線程模型對我們來説尤其重要。

8.1. Spring Data MongoDB

Spring Data MongoDB 提供基於 MongoDB Reactive Streams 驅動程序的反應式存儲庫支持。 尤其值得注意的是，該驅動程序完全實現了 Reactive Streams API，以提供異步流處理和非阻塞背壓。

在 Spring Boot 應用程序中設置 MongoDB 反應式存儲庫支持，只需添加一個依賴項即可：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>

這將使我們能夠創建一個存儲庫，並使用它以非阻塞的方式對 MongoDB 執行一些基本操作：

public interface PersonRepository extends ReactiveMongoRepository<Person, ObjectId> {
}
.....
personRepository.findAll().doOnComplete(this::printThreads);

我們能看到什麼類型的線程當我們在 Netty 服務器上運行這個應用程序呢？

當然，正如預料的那樣，由於 一個 Spring Data 反應式倉庫 利用了與服務器相同的事件循環，因此不會有太大的差異。

8.2. Reactor Kafka

Spring 仍在積極構建對 Reactive Kafka 的全面支持。 不過，我們也有一些在 Spring 之外的選項。

Reactor Kafka 是基於 Reactor 的 Reactive API，用於 Kafka。Reactor Kafka 允許使用函數式 API 發佈和消費消息，並具有 非阻塞背壓。

首先，我們需要在應用程序中添加所需的依賴項，才能開始使用 Reactor Kafka：

<dependency>
    <groupId>io.projectreactor.kafka</groupId>
    <artifactId>reactor-kafka</artifactId>
    <version>1.3.10</version>
</dependency>

這將使我們能夠以非阻塞的方式向 Kafka 生產消息：

// producerProps: Map of Standard Kafka Producer Configurations
SenderOptions<Integer, String> senderOptions = SenderOptions.create(producerProps);
KafkaSender<Integer, String> sender =  KafkaSender.create(senderOptions);
Flux<SenderRecord<Integer, String, Integer>> outboundFlux = Flux
  .range(1, 10)
  .map(i -> SenderRecord.create(new ProducerRecord<>("reactive-test", i, "Message_" + i), i));
sender.send(outboundFlux).subscribe();

同樣，我們應該能夠以非阻塞的方式消費來自 Kafka 的消息：

// consumerProps: Map of Standard Kafka Consumer Configurations
ReceiverOptions<Integer, String> receiverOptions = ReceiverOptions.create(consumerProps);
receiverOptions.subscription(Collections.singleton("reactive-test"));
KafkaReceiver<Integer, String> receiver = KafkaReceiver.create(receiverOptions);
Flux<ReceiverRecord<Integer, String>> inboundFlux = receiver.receive();
inboundFlux.doOnComplete(this::printThreads)

這非常簡單且自解釋。

我們訂閲了 Kafka 中的主題 reactive-test，並接收到一條 Flux 消息流。

對我們來説最有趣的是創建的線程：

我們可以看到一些與 Netty 服務器不常見的線程。

這表明 Reactor Kafka 自身管理線程池，其中包含一些參與 Kafka 消息處理的工進程。當然，我們還會看到與 Netty 和 JVM 相關的其他線程，這些線程可以忽略不計。

Kafka 生產者使用單獨的網絡線程來向 broker 發送請求。 此外，它們使用一個 單線程池調度器 將響應傳遞給應用程序。

Kafka 消費者，另一方面，為每個消費者組都有一個線程，該線程會阻塞以監聽傳入的消息。 傳入的消息隨後會安排在另一個線程池中進行處理。

9. WebFlux 中的調度選項

我們已經瞭解到，在完全非阻塞的環境中，僅使用少量線程，反應式編程真正發光發熱。但這同時也意味着，如果確實存在阻塞的部分，它會導致性能大幅下降。這是因為阻塞操作會完全凍結事件循環。

因此，我們如何處理反應式編程中的長時間運行進程或阻塞操作？

誠然，最佳方案是儘量避免它們。然而，這並非總是可行的，我們可能需要為應用程序中的某些部分制定專門的調度策略。

Spring WebFlux 提供了一種機制，可以在數據流鏈之間切換到不同的線程池。這使我們能夠對某些任務的調度策略進行精確控制。當然，WebFlux 能夠基於底層反應式庫中提供的調度器（即線程池抽象）來實現這一點。

9.1. 反應器 (Reactor)

在 反應器 (Reactor) 中，<strong><em>調度器 (Scheduler)</em> 類定義了執行模型，以及執行的地點。

<a href="https://projectreactor.io/docs/core/release/api/reactor/core/scheduler/Schedulers.html">調度器 (Schedulers)</strong> 類提供了多種執行上下文，例如 <em>immediate</em>、<em>single</em>、<em>elastic</em> 和 <em>parallel</em>。 這些提供了不同類型的線程池，對於不同的任務很有用。 此外，我們始終可以創建一個具有預現有 ExecutorService 的自定義 Scheduler。

雖然 <em>Scheduler</em> 提供了多種執行上下文，但反應器 <strong>還提供了不同的切換執行上下文的方式</strong>。 這些是publishOn和subscribeOn` 方法。

我們可以使用 <em>publishOn</em> 與鏈中的任何 <em>Scheduler</em> 結合使用，該 <em>Scheduler</em> 會影響所有後續操作符。

雖然我們也可以使用 <em>subscribeOn</em> 與鏈中的任何 <em>Scheduler</em> 結合使用，但它只會影響發射源的上下文。

如果回想一下，WebClient 在 Netty 上共享了服務器端創建的相同 事件循環 (event loop) 作為默認行為。 但是，我們可能出於有充分理由創建 WebClient 的專用線程池。

Scheduler scheduler = Schedulers.newBoundedElastic(5, 10, "MyThreadGroup");

WebClient.create("http://localhost:8080/index").get()
  .retrieve()
  .bodyToMono(String.class)
  .publishOn(scheduler)
  .doOnNext(s -> printThreads());

此前，我們並未觀察到在 Netty 中使用或不使用 WebClient 時線程創建方面存在差異。但是，如果現在運行上述代碼，我們將會觀察到一些新的線程被創建：

在這裏 我們可以看到這些線程是作為我們 有限彈性線程池 的一部分創建的。這是 WebClient 的響應發佈在此處的環節。

這使得主線程池可以用於處理服務器請求。

9.2. RxJava

默認情況下，RxJava的行為與Reactor非常相似。

Observable及其上應用的所有運算符將在訂閲被調用時執行工作並通知觀察者。 就像Reactor一樣，RxJava也提供了一種在鏈中引入前綴或自定義調度策略的方式。

RxJava還包含一個類<a href="http://reactivex.io/RxJava/javadoc/io.reactivex.schedulers.Schedulers.html">Schedulers</a>，該類提供了對<a href="http://reactivex.io/RxJava/javadoc/io.reactivex.Observable.html"><em>Observable</em></a>鏈的多種執行模型。 這些模型包括<em>new thread</em>、<em>immediate</em>、<em>trampoline</em>、<em>io</em>、<em>computation</em>和<em>test</em>。 當然，它還允許我們定義一個<a href="http://reactivex.io/documentation/scheduler.html">Scheduler</a>，該Scheduler來自Java的<a href="https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/Executor.html"><em>Executor</em></a>。

此外，RxJava還提供兩個擴展方法來實現這一點，即<em>subscribeOn</em>和<em>observeOn</em>。

<em>subscribeOn</em>方法通過指定<em>Observable</em>應該運行的不同的<em>Scheduler</em>來改變默認行為。 另一方面，<em>observeOn</em>方法指定<em>Observable</em>可以使用該<em>Scheduler</em>向觀察者發送通知。

正如我們之前討論的，Spring WebFlux默認使用Reactor作為其響應式庫。 但是，由於它與Reactive Streams API完全兼容，因此可以切換到另一個Reactive Streams實現，例如RxJava（對於RxJava 1.x及其Reactive Streams適配器）。

我們需要顯式添加依賴：

<dependency>
    <groupId>io.reactivex.rxjava2</groupId>
    <artifactId>rxjava</artifactId>
    <version>2.2.21</version>
</dependency>

然後我們可以開始在我們的應用程序中使用 RxJava 類型，例如 Observable，以及 RxJava 相關的 調度器：

io.reactivex.Observable
  .fromIterable(Arrays.asList("Tom", "Sawyer"))
  .map(s -> s.toUpperCase())
  .observeOn(io.reactivex.schedulers.Schedulers.trampoline())
  .doOnComplete(this::printThreads);

因此，如果運行該應用程序，除了常規的 Netty 和 JVM 相關的線程之外，我們應該看到一些與 RxJava 調度器相關的線程：

10. 結論

在本文中，我們從併發的角度探討了響應式編程的 premises。我們觀察到傳統響應式編程與併發模型之間的差異，從而得以研究 Spring WebFlux 中的併發模型及其實現方式。

隨後，我們結合了 WebFlux 中的線程模型與不同的 HTTP 運行時和響應式庫，並學習了使用 WebClient 與使用數據訪問庫時線程模型之間的差異。

最後，我們探討了在 WebFlux 中控制程序調度策略的選項。