響應式編程簡介之如何掌握Reactor

這篇文章主要介紹“響應式編程簡介之如何掌握Reactor”，在日常操作中，相信很多人在響應式編程簡介之如何掌握Reactor問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”響應式編程簡介之如何掌握Reactor”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

Reactor簡介

Reactor是基于JVM的非阻塞API，他直接跟JDK8中的API相結合，比如：CompletableFuture，Stream和Duration等。

它提供了兩個非常有用的異步序列API：Flux和Mono，并且實現了Reactive Streams的標準。

并且還可以和reactor-netty相結合，作為一些異步框架的底層服務，比如我們非常熟悉的Spring MVC 5中引入的WebFlux。

我們知道WebFlux的底層使用的是reactor-netty，而reactor-netty又引用了Reactor。所以，如果你在POM中引入了webFlux依賴：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

那么項目將會自動引入Reactor。

如果你用的不是Spring webflux，沒關系，你可以直接添加下面的依賴來使用Reactor：

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
</dependency>

reactive programming的發展史

最最開始的時候微軟為.NET平臺創建了Reactive Extensions (Rx) library。接著RxJava實現了JVM平臺的Reactive。

然后Reactive Streams標準出現了，它定義了Java平臺必須滿足的的一些規范。并且已經集成到JDK9中的java.util.concurrent類中。

在Flow中定義了實現Reactive Streams的四個非常重要的組件，分別是Publisher，Subscriber，Subscription和Processor。

Iterable-Iterator 和Publisher-Subscriber的區別

一般來說reactive在面向對象的編程語言中是以觀察者模式的擴展來使用的。

我們來具體看一下這個觀察者模式的實現，以Publisher和Subscriber為例：

   public static interface Publisher<T> {
        public void subscribe(Subscriber<? super T> subscriber);
    }

    public static interface Subscriber<T> {

        public void onSubscribe(Subscription subscription);

        public void onNext(T item);

        public void onError(Throwable throwable);

        public void onComplete();
    }

上面定義了兩個接口，Publisher和Subscriber，Publisher的作用就是subscribe到subscriber。

而subscriber定義了4個on方法，用來觸發特定的事件。

那么Publisher中的subscribe是怎么觸發Subscriber的onSubscribe事件呢？

很簡單，我們看一個具體的實現：

    public void subscribe(Flow.Subscriber<? super T> subscriber) {
        Subscription sub;
        if (throwable != null) {
            assert iterable == null : "non-null iterable: " + iterable;
            sub = new Subscription(subscriber, null, throwable);
        } else {
            assert throwable == null : "non-null exception: " + throwable;
            sub = new Subscription(subscriber, iterable.iterator(), null);
        }
        subscriber.onSubscribe(sub);

        if (throwable != null) {
            sub.pullScheduler.runOrSchedule();
        }
    }

上面的例子是PullPublisher的subscribe實現。我們可以看到，在這個subscribe中觸發了subscriber.onSubscribe方法。而這就是觀察者模式的秘密。

或者說，當Publisher調用subscribe的時候，是主動push subscriber的onSubscribe方法。

熟悉Iterable-Iterator模式的朋友應該都知道，Iterator模式，其實是一個主動的pull模式，因為需要不斷的去調用next()方法。所以它的控制權是在調用方。

為什么要使用異步reactive

在現代應用程序中，隨著用戶量的增多，程序員需要考慮怎么才能提升系統的處理能力。

傳統的block IO的方式，因為需要占用大量的資源，所以是不適合這樣的場景的。我們需要的是NO-block IO。

JDK中提供了兩種異步編程的模型：

第一種是Callbacks，異步方法可以通過傳入一個Callback參數的形式來在Callback中執行異步任務。比較典型的像是java Swing中的EventListener。

第二中就是使用Future了。我們使用Callable來提交一個任務，然后通過Future來拿到它的運行結果。

這兩種異步編程會有什么問題呢？

callback的問題就在于回調地獄。熟悉JS的朋友應該很理解這個回調地獄的概念。

簡單點講，回調地獄就是在callback中又使用了callback，從而造成了這種callback的層級調用關系。

而Future主要是對一個異步執行的結果進行獲取，它的 get()實際上是一個block操作。并且不支持異常處理，也不支持延遲計算。

當有多個Future的組合應該怎么處理呢？JDK8 實際上引入了一個CompletableFuture類，這個類是Future也是一個CompletionStage，CompletableFuture支持then的級聯操作。不過CompletableFuture提供的方法不是那么的豐富，可能滿足不了我的需求。

于是我們的Reactor來了。

Flux

Reactor提供了兩個非常有用的操作，他們是 Flux 和 Mono。 其中Flux 代表的是 0 to N 個響應式序列，而Mono代表的是0或者1個響應式序列。

我們看一個Flux是怎么transfer items的：

先看下Flux的定義：

public abstract class Flux<T> implements Publisher<T>

可以看到Flux其實就是一個Publisher，用來產生異步序列。

Flux提供了非常多的有用的方法，來處理這些序列，并且提供了completion和error的信號通知。

相應的會去調用Subscriber的onNext, onComplete, 和 onError 方法。

Mono

我們看下Mono是怎么transfer items的：

看下Mono的定義：

public abstract class Mono<T> implements Publisher<T>

Mono和Flux一樣，也是一個Publisher，用來產生異步序列。

Mono因為只有0或者1個序列，所以只會觸發Subscriber的onComplete和onError方法，沒有onNext。

另一方面，Mono其實可以看做Flux的子集，只包含Flux的部分功能。

Mono和Flux是可以互相轉換的，比如Mono#concatWith(Publisher)返回一個Flux，而 Mono#then(Mono)返回一個Mono.

Flux和Mono的基本操作

我們看下Flux創建的例子：

Flux<String> seq1 = Flux.just("foo", "bar", "foobar");
List<String> iterable = Arrays.asList("foo", "bar", "foobar");
Flux<String> seq2 = Flux.fromIterable(iterable);
Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);

可以看到Flux提供了很多種創建的方式，我們可以自由選擇。

再看看Flux的subscribe方法：

Disposable subscribe(); 

Disposable subscribe(Consumer<? super T> consumer); 

Disposable subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer); 

Disposable subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer); 

Disposable subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer,
          Consumer<? super Subscription> subscriptionConsumer);

subscribe可以一個參數都沒有，也可以多達4個參數。

看下沒有參數的情況:

Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);

numbersFromFiveToSeven.subscribe();

注意，沒有參數并不表示Flux的對象不被消費，只是不可見而已。

看下帶參數的情況：consumer用來處理on each事件，errorConsumer用來處理on error事件，completeConsumer用來處理on complete事件，subscriptionConsumer用來處理on subscribe事件。

前面的3個參數很好理解，我們來舉個例子：

Flux<Integer> ints3 = Flux.range(1, 4);
        ints3.subscribe(System.out::println,
                error -> System.err.println("Error " + error),
                () -> System.out.println("Done"),
                sub -> sub.request(2));

我們構建了從1到4的四個整數的Flux，on each就是打印出來，如果中間有錯誤的話，就輸出Error，全部完成就輸出Done。

那么最后一個subscriptionConsumer是做什么用的呢？

subscriptionConsumer accept的是一個Subscription對象，我們看下Subscription的定義：

public interface Subscription {

    public void request(long n);
    public void cancel();
}

Subscription 定義了兩個方法，用來做初始化用的，我們可以調用request(n)來決定這次subscribe獲取元素的最大數目。

比如上面我們的例子中，雖然構建了4個整數，但是最終輸出的只有2個。

上面所有的subscribe方法，都會返回一個Disposable對象，我們可以通過Disposable對象的dispose()方法，來取消這個subscribe。

Disposable只定義了兩個方法：

public interface Disposable {

	void dispose();

	default boolean isDisposed() {
		return false;
	}

dispose的原理是向Flux 或者 Mono發出一個停止產生新對象的信號，但是并不能保證對象產生馬上停止。

有了Disposable，當然要介紹它的工具類Disposables。

Disposables.swap() 可以創建一個Disposable，用來替換或者取消一個現有的Disposable。

Disposables.composite(…)可以將多個Disposable合并起來，在后面統一做處理。

到此，關于“響應式編程簡介之如何掌握Reactor”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！