深入探究Project Reactor

一、Project Reactor 簡介

// 以下是 import 部分
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;

public class ReactorDemo {
    public static void main(String[] args) throws InterruptedException {
        Flux flux = Flux.just("Hello", "World", "Reactor");
        flux.subscribe(System.out::println);
    }
}

Project Reactor 是一個基於 Reactive Stream API 之上的 reactive 庫。Reactive Stream 是一個規範，用於通訊無阻塞背壓非同步流的交互。Project Reactor 實現了 Reactive Stream API 規範，同時針對 Java 8 的特性也進行了進一步的擴展，支持 Reactor 的 Mono 和 Flux API，讓使用者能夠方便地使用 Java 8 的 Stream 相關 API 進行數據處理。

Reactor 的核心構建模塊為 Mono 和 Flux，該模塊提供了類似於Java Stream 的編程模型，支持生成、處理序列數據。其中 Flux 表示的是包含 0 到 N 個數據元素的非同步序列，而 Mono 則表示包含 0 或者 1 個數據元素的非同步序列，也就是可以返回數據也可以返回值為空，同時在底層實現上，這兩個類也提供了流式操作 API，例如：map、filter、reduce 等。

以下是 Flux 的示例代碼：

Flux.just("Hello", "World", "Reactor")
    .map(s -> s.concat(", "))
    .zipWith(Flux.range(1, Integer.MAX_VALUE), (s, index) -> index + ". " + s)
    .subscribe(System.out::print);
// Output: 1. Hello, 2. World, 3. Reactor, 

在上述代碼中，我們先通過 Flux.just 初始化了一個有三個元素的 Flux 對象，接著使用 map 進行字元串拼接，通過 zipWith 將兩個 Flux 對象先後對應地組合在一起，最後通過 subscribe 方法將結果輸出。

二、Project Reactor 編程模型

1. Flux 模型

Flux 表示包含 0 到 N 個數據元素的非同步序列，可以用於推送數據到一個訂閱者流中。Flux 支持兩種方式的數據推送：訂閱和非同步執行。當數據源被訂閱時，Flux 將按照訂閱者指定的方式推送數據。當沒有訂閱者時，數據源不會進行推送。非同步執行時，Flux 採用 Reactor 提供的 Schedulers 線程調度進行非同步處理。

以下是 Flux 的基本用法：

// 以下是 import 部分
import reactor.core.publisher.Flux;

public class FluxDemo {
    public static void main(String[] args) {
        Flux.just("Hello", "World", "Reactor")
            .subscribe(System.out::println);
    }
}

這段代碼演示了如何在控制台列印 Flux 數據流中的元素。在 subscribe 方法執行後，Flux 將立即開始推送在 just 方法中指定的數據。

2. Mono 模型

Mono 則表示一個包含 0 或 1 個數據元素的非同步序列，同樣也可以用於推送數據到一個訂閱者流中。它和 Flux 唯一的區別在於數據元素的數量。

以下是 Mono 的基本用法：

// 以下是 import 部分
import reactor.core.publisher.Mono;

public class MonoDemo {
    public static void main(String[] args) {
        Mono.just("Hello Reactor")
            .subscribe(System.out::println);
    }
}

這段代碼演示了如何在控制台列印 Mono 數據流中的元素。在 subscribe 方法執行後，Mono 將立即開始推送在 just 方法中指定的數據。

三、Project Reactor 批處理

當需要對數據流進行批處理時，可以使用 buffer 操作符將多個元素合併到一個 List 中，形成一個批次，然後把這個 List 推給下游流。如下所示：

// 以下是 import 部分
import reactor.core.publisher.Flux;

public class BufferDemo {
    public static void main(String[] args) {
        Flux.range(1, 10)
            .buffer(3)
            .subscribe(System.out::println);
    }
}

這個例子的中 buffer(3) 操作符創建了一個可以包含 3 個元素的 List。Flux 已經發出了 10 個元素，因此在推送完所有元素之前，將生成三個 List。輸出如下：

[1, 2, 3]
[4, 5, 6]
[7, 8, 9]
[10]

四、Project Reactor 選擇線程

Reactor 提供了眾多的線程池支持，以便於讓開發者可以自定義線程池，通過配置不同的線程池來實現多線程編程。以下是官網提供的線程選擇策略，針對不同的場景可以選擇不同的策略：

• boundedElastic：一個支持生命周期的線程池（回收線程），默認可以使用 Integer.MAX_VALUE 個線程，其中每個線程都是預定義的，以保證線程可以在幾納秒內就可以被重用。此線程池適用於執行計算密集型任務，如單獨線程的映射，同步 IO 或 CPU 綁定的阻塞。此線程池不支持任務隊列，每次提交任務都會立即執行。
• elastic：使用無界隊列的線程池，每次提交任務都會創建一個新的線程。此線程池適用於執行 CPU 密集型任務，因為這樣可以保證線程不會被阻塞。
• single：使用一個線程處理任務，適用於執行多個任務中的串列計算，並使得任務在相同的線程上運行。此線程池適用於執行串列計算，如 Stream -> Stream。
• immediate：直接在訂閱線程上執行任務。

以下是使用 elastic 線程池的代碼示例：

// 以下是 import 部分
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class SchedulerDemo {
    public static void main(String[] args) {
        Flux.just("Hello", "World", "Reactor")
            .publishOn(Schedulers.elastic())
            .map(String::toUpperCase)
            .subscribe(System.out::println);
    }
}

在上述示例中，通過 publishOn 方法選擇 Schedulers.elastic() 線程池，使用其線程池進行數據處理，然後通過 map 方法將數據進行處理、轉化為大寫，最後輸出結果。

五、Project Reactor 反應式API

Project Reactor 提供了 Flux 和 Mono 類，這些類提供了一組反應式的 API，用於處理非同步數據流，並提供了相關的方法進行數據處理。

以下代碼示例使用反應式 API 計算 Stream 數據的平均值：

// 以下是 import 部分
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.stream.Stream;

public class ReactiveAPIDemo {
    public static void main(String[] args) {
        Flux intervalFlux = Flux.interval(Duration.ofSeconds(1));
        Stream inputStream = Stream.of(5.0f, 6.0f, 7.0f);
        Flux.fromStream(inputStream)
                .zipWith(intervalFlux, (i, t) -> i)
                .buffer(3)
                .map(buf -> {
                    float sum = buf.stream().reduce(Float::sum).orElse(0f);
                    return sum / buf.size();
                })
                .subscribe(System.out::println);
    }
}

在上面的代碼示例中我們首先創建了自動發出遞增長整數序列的 Flux，接著使用 zipWith 操作符將其和 Stream 數據進行組合，每 3 個組成一組。最後，我們使用 map 函數計算組的平均值，並將組對象返回。最終輸出結果如下：

6.0

六、結語

通過本文介紹，我們已經學習了 Project Reactor 的核心構建模塊 Flux 和 Mono 的使用方法，以及 Project Reactor 的編程模型和批處理。通過調整選擇線程池的位置可以控制代碼的並發程度，提高代碼的處理效率。同時通過之前的示例我們也學習了反應式 API 的使用，我們可以在項目中通過這個 API 實現多線程非同步計算，優化項目性能，提高代碼質量和可讀性。