2. 程式人生 > >（11）照虎畫貓深入理解響應式流規範——響應式Spring的道法術器

（11）照虎畫貓深入理解響應式流規範——響應式Spring的道法術器

阿新 • • 發佈：2019-01-10

2 響應式程式設計之法

上一章本著“快速上手”的原則，介紹了響應式流的概念，以及Reactor 3的使用。這一章，我們基於Reactor 3的實現原理，從《響應式流規範》入手，深入瞭解響應式流開發庫。

2.1 響應式流規範

現代軟體對近乎實時地處理資料的需求越來越強烈，對不斷變化的資訊的即時響應，意味著更大的商業價值，流處理是一種快速將資料轉換為有用資訊的手段。

資料流中的元素可以是一個一個的待計算的資料，也可以是一個一個待響應的事件。前者多用於大資料處理，比如Storm、Spark等產品，後者常用於響應式程式設計，比如Netflix在使用的RxJava、Scala程式語言的發明者Typesafe公司（已更名為Lightbend）的Akka Stream、Java開發者都熟悉的Pivotal公司的Project Reactor、走在技術前沿的Vert.x等。

軟體行業是一個非常注重分享和交流的行業。隨著對響應式程式設計技術的討論與溝通逐漸深入，2013年末的時候，Netflix、Pivotal、Typesafe等公司的工程師們共同發起了關於制定“響應式流規範（Reactive Stream Specification）”的倡議和討論，並在github上建立了reactive-streams-jvm專案。到2015年5月份，1.0版本的規範出爐，專案README就是規範正文。

各個響應式開發庫都要遵循這個規範，其好處也是顯而易見的。之所以我們編寫的Java程式碼可以在Hotspot、J9和Zing等JVM執行，是因為它們都遵循Java虛擬機器規範。類似的，由於各個響應式開發庫都遵循響應式流規範，因此互相相容，不同的開發庫之間可以進行互動，我們甚至可以同時在專案中使用多個響應式開發庫。對於Spring WebFlux來說，也可以使用RxJava作為響應式庫。

雖然響應式流規範是用來約束響應式開發庫的，作為使用者的我們如果能夠了解這一規範對於我們理解開發庫的使用也是很有幫助的，因為規範的內容都是對響應式程式設計思想的精髓的呈現。訪問reactive-streams-jvm專案，可以瀏覽規範的細節，包括其中定義的響應式流的特點：

  1. 具有處理無限數量的元素的能力；
  2. 按序處理；
  3. 非同步地傳遞元素；
  4. 必須實現非阻塞的回壓（backpressure）。

2.1.1 響應式流介面

響應式流規範定義了四個介面，如下：

1.Publisher是能夠發出元素的釋出者。

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

2.Subscriber是接收元素並做出響應的訂閱者。

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

當執行subscribe方法時，釋出者會回撥訂閱者的onSubscribe方法，這個方法中，通常訂閱者會藉助傳入的Subscription向釋出者請求n個數據。然後釋出者通過不斷呼叫訂閱者的onNext方法向訂閱者發出最多n個數據。如果資料全部發完，則會呼叫onComplete告知訂閱者流已經發完；如果有錯誤發生，則通過onError發出錯誤資料，同樣也會終止流。

title

訂閱後的回撥用表示式表示就是onSubscribe onNext* (onError | onComplete)?，即以一個onSubscribe開始，中間有0個或多個onNext，最後有0個或1個onError或onComplete事件。

Publisher和Subscriber融合了迭代器模式和觀察者模式。

我們經常用到的Iterable和Iterator就是迭代器模式的體現，可以滿足上邊第1和2個特點關於按需處理資料流的要求；而觀察者模式基於事件的回撥機制有助於滿足第3個特點關於非同步傳遞元素的要求。

3.Subscription是Publisher和Subscriber的“中間人”。

public interface Subscription {
    public void request(long n);
    public void cancel();
}

當釋出者呼叫subscribe方法註冊訂閱者時，會通過訂閱者的回撥方法onSubscribe傳入Subscription物件，之後訂閱者就可以使用這個Subscription物件的request方法向釋出者“要”資料了。回壓機制正是基於此來實現的，因此第4個特點也能夠實現了。

4.Processor集Publisher和Subscriber於一身。

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

這四個介面是實現各開發庫之間互相相容的橋樑，響應式流規範也僅僅聚焦於此，而對諸如轉換、合併、分組等等的操作一概未做要求，因此是一個非常抽象且精簡的介面規範。

如果這時候有人要造輪子，再寫一套響應式開發庫，如何基於這幾個介面展開呢？

2.1.2 照虎畫貓，理解訂閱後發生了什麼

Reactor 3是遵循響應式流規範的實現，因此，小擼一把Reactor的原始碼有助於我們理解規範中定義的介面的使用。

Reactor中，我們最先接觸的生成Publisher的方法就是Flux.just()，下面我們來動手寫程式碼模擬一下Reactor的實現方式。不過具備生產能力開發庫會考慮效能、併發安全性等諸多因素，所謂“照虎畫貓”，我們的程式碼只是模擬出實現思路，程式碼少的多，但五臟俱全。

    <dependency>
        <groupId>org.reactivestreams</groupId>
        <artifactId>reactive-streams</artifactId>
        <version>1.0.2</version>
    </dependency>

首先建立最最基礎的類Flux，它是一個Publisher。

package reactor.core.publisher;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

public abstract class Flux<T> implements Publisher<T> {
    public abstract void subscribe(Subscriber<? super T> s);
}

在Reactor中，Flux既是一個釋出者，又充當工具類的角色，當我們用Flux.just()、Flux.range()或Flux.interval()等工廠方法生成Flux時，會new一個新的Flux，比如Flux.just會返回一個FluxArray物件。

    public static <T> Flux<T> just(T... data) {
        return new FluxArray<>(data);
    }

返回的FluxArray物件是Flux.just生成的Publisher，它繼承自Flux，並實現了subscribe方法。

public class FluxArray<T> extends Flux<T> {
    private T[] array;  // 1

    public FluxArray(T[] data) {
        this.array = data;
    }

    @Override
    public void subscribe(Subscriber<? super T> actual) {
        actual.onSubscribe(new ArraySubscription<>(actual, array)); // 2
    }
}
  1. FluxArray內部使用一個數組來儲存資料；
  2. subscribe方法通常會回撥Subscriber的onSubscribe方法，該方法需要傳入一個Subscription物件，從而訂閱者之後可以通過回撥傳回的Subscription的request方法跟FluxArray請求資料，這也是回壓的應有之義。

繼續編寫ArraySubscription：

public class FluxArray<T> extends Flux<T> {
    ...
    static class ArraySubscription<T> implements Subscription { // 1
        final Subscriber<? super T> actual;
        final T[] array;    // 2
        int index;
        boolean canceled;

        public ArraySubscription(Subscriber<? super T> actual, T[] array) {
            this.actual = actual;
            this.array = array;
        }

        @Override
        public void request(long n) {
            if (canceled) {
                return;
            }
            long length = array.length;
            for (int i = 0; i < n && index < length; i++) {
                actual.onNext(array[index++]);  // 3
            }
            if (index == length) {
                actual.onComplete();    // 4
            }
        }

        @Override
        public void cancel() {  // 5
            this.canceled = true;
        }
    }
}
  1. ArraySubscription是一個靜態內部類。靜態內部類是最簡單的一種內部類，你儘可以把它當成普通的類，只不過恰好定義在其他類的內部；
  2. 可見在Subscription內也有一份資料；
  3. 當有可以發出的元素時，回撥訂閱者的onNext方法傳遞元素；
  4. 當所有的元素都發完時，回撥訂閱者的onComplete方法；
  5. 訂閱者可以使用Subscription取消訂閱。

到此為止，釋出者就開發完了。我們測試一下：

@Test
public void fluxArrayTest() {
    Flux.just(1, 2, 3, 4, 5).subscribe(new Subscriber<Integer>() { // 1

        @Override
        public void onSubscribe(Subscription s) {
            System.out.println("onSubscribe");
            s.request(6);   // 2
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println("onNext:" + integer);
        }

        @Override
        public void onError(Throwable t) {

        }

        @Override
        public void onComplete() {
            System.out.println("onComplete");
        }
    });
}
  1. Subscriber通過匿名內部類定義，其中需要實現介面的四個方法；
  2. 訂閱時請求6個元素。

測試方法執行如下：

1
2
3
4
5
Completed.

如果請求3個元素呢？輸出如下：

1
2
3

沒有完成事件，OK，一個簡單的Flux.just就完成了，通過這個例子我們能夠初步摸出Flux工廠方法的一些“套路”：

  • 工廠方法返回的是Flux子類的例項，如FluxArray；
  • FluxArray的subscribe方法會返回給訂閱者一個Subscription實現類的物件，這個ArraySubscription是FluxArray的靜態內部類，定義了“如何釋出元素”的邏輯；
  • 訂閱者可以通過這個ArraySubscription物件向釋出者請求n個數據；釋出者也可以藉助這個ArraySubscription物件向訂閱者傳遞資料元素（onNext/onError/onComplete）。

用圖來表示如下（由於Subscription是靜態內部類，可以看做普通類，就單獨放一邊了）：

title

上圖的這個過程基本適用於大多數的用於生成Flux/Mono的靜態工廠方法，如Flux.just、Flux.range等。

首先，使用類似Flux.just的方法建立釋出者後，會建立一個具體的釋出者（Publisher），如FluxArray。

  1. 當使用.subscribe訂閱這個釋出者時，首先會new一個具有相應邏輯的Subscription（如ArraySubscription，這個Subscription定義瞭如何處理下游的request，以及如何“發出資料”）；
  2. 然後釋出者將這個Subscription通過訂閱者的.onSubscribe方法傳給訂閱者；
  3. 在訂閱者的.onSubscribe方法中，需要通過Subscription發起第一次的請求.request；
  4. Subscription收到請求，就可以通過回撥訂閱者的onNext方法發出元素了，有多少發多少，但不能超過請求的個數；
  5. 訂閱者在onNext中通常定義對元素的處理邏輯，處理完成之後，可以繼續發起請求；
  6. 釋出者根據繼續滿足訂閱者的請求；
  7. 直至釋出者的序列結束，通過訂閱者的onComplete予以告知；當然序列傳送過程中如果有錯誤，則通過訂閱者的onError予以告知並傳遞錯誤資訊；這兩種情況都會導致序列終止，訂閱過程結束。

以上從1~7這些階段稱為訂閱期（subscribe time）。

2.1.3 照虎畫貓——操作符“流水線”

響應式開發庫的一個很讚的特性就是可以像組裝流水線一樣將操作符串起來，用來聲明覆雜的處理邏輯。比如：

Flux ff = Flux.just(1, 2, 3, 4, 5)
    .map(i -> i * i)
    .filter(i -> (i % 2) == 0);
ff.subscribe(...)

通過原始碼，我們可以瞭解這種“流水線”的實現機制。下面我們仍然是通過照虎畫貓的方式模擬一下Reactor中Flux.map的實現方式。

Flux.map用於實現轉換，轉換後元素的型別可能會發生變化，轉換的邏輯由引數Function決定。方法本身返回的是一個轉換後的Flux，基於此，該方法實現如下：

public abstract class Flux<T> implements Publisher<T> {
    ...
    public <V> Flux<V> map(Function<? super T, ? extends V> mapper) {   // 1
        return new FluxMap<>(this, mapper); // 2
    }
}
  1. 泛型方法，通過泛型表示可能出現的型別的變化（T → V）；
  2. FluxMap就是新的Flux。

既然FluxMap是一個新的Flux，那麼與2.1.2中FluxArray類似，其內部定義有MapSubscription，這是一個Subscription，能夠根據其訂閱者的請求發出資料。

public class FluxMap<T, R> extends Flux<R> {

    private final Flux<? extends T> source;
    private final Function<? super T, ? extends R> mapper;

    public FluxMap(Flux<? extends T> source, Function<? super T, ? extends R> mapper) {
        this.source = source;
        this.mapper = mapper;
    }

    @Override
    public void subscribe(Subscriber<? super R> actual) {
        source.subscribe(new MapSubscriber<>(actual, mapper));
    }

    static final class MapSubscription<T, R> implements Subscription {
        private final Subscriber<? super R> actual;
        private final Function<? super T, ? extends R> mapper;

        MapSubscriber(Subscriber<? super R> actual, Function<? super T, ? extends R> mapper) {
            this.actual = actual;
            this.mapper = mapper;
        }

        @Override
        public void request(long n) {   // 1
            // TODO 收到請求，發出元素
        }

        @Override
        public void cancel() {
            // TODO 取消訂閱
        }
    }
}
  1. 但是map操作符並不產生資料，只是資料的搬運工。收到request後要發出的資料來自上游。

所以MapSubscription同時也應該是一個訂閱者，它訂閱上游的釋出者，並將資料處理後傳遞給下游的訂閱者（為了跟Reactor原始碼一致，將MapSubscription改名為MapSubscriber，其實沒差）。

title

如圖，對下游是作為釋出者，傳遞上游的資料到下游；對上游是作為訂閱者，傳遞下游的請求到上游。

static final class MapSubscriber<T, R> implements Subscriber<T>, Subscription { // 1
    ...
}
  1. 實現了Subscriber和Subscription。

這樣，總共有5個方法要實現：來自Subscriber介面的onSubscribe、onNext、onError、onComplete，和來自Subscription介面的request和cancel。下面我們本著“搬運工”的角色實現這幾個方法即可。

static final class MapSubscriber<T, R> implements Subscriber<T>, Subscription {
    private final Subscriber<? super R> actual;
    private final Function<? super T, ? extends R> mapper;

    boolean done;

    Subscription subscriptionOfUpstream;

    MapSubscriber(Subscriber<? super R> actual, Function<? super T, ? extends R> mapper) {
        this.actual = actual;
        this.mapper = mapper;
    }

    @Override
    public void onSubscribe(Subscription s) {
        this.subscriptionOfUpstream = s;    // 1
        actual.onSubscribe(this);           // 2
    }

    @Override
    public void onNext(T t) {
        if (done) {
            return;
        }
        actual.onNext(mapper.apply(t));     // 3
    }

    @Override
    public void onError(Throwable t) { 
        if (done) {
            return;
        }
        done = true;
        actual.onError(t);                  // 4
    }

    @Override
    public void onComplete() {
        if (done) {
            return;
        }
        done = true;
        actual.onComplete();                // 5
    }

    @Override
    public void request(long n) {
        this.subscriptionOfUpstream.request(n);     // 6
    }

    @Override
    public void cancel() {
        this.subscriptionOfUpstream.cancel();       // 7
    }
}
  1. 拿到來自上游的Subscription；
  2. 回撥下游的onSubscribe，將自身作為Subscription傳遞過去；
  3. 收到上游發出的資料後，將其用mapper進行轉換，然後接著發給下游；
  4. 將上游的錯誤訊號原樣發給下游；
  5. 將上游的完成訊號原樣發給下游；
  6. 將下游的請求傳遞給上游；
  7. 將下游的取消操作傳遞給上游。

從這個對原始碼的模仿，可以體會到，當有多個操作符串成“操作鏈”的時候：

  • 向下：很自然地，資料和訊號（onSubscribe、onNext、onError、onComplete）是通過每一個操作符向下傳遞的，傳遞的過程中進行相應的操作處理，這一點並不難理解；
  • 向上：然而在內部我們看不到的是，有一個自下而上的“訂閱鏈”，這個訂閱鏈可以用來傳遞request，因此回壓（backpressure）可以實現從下游向上遊的傳遞。

這一節最開頭的那一段程式碼的執行過程如下圖所示：

title

2.1.4 LambdaSubscriber

在1.3.2節的時候，介紹了.subscribe的幾個不同方法簽名的變種：

subscribe(  Consumer<? super T> consumer) 

subscribe(  @Nullable Consumer<? super T> consumer, 
            Consumer<? super Throwable> errorConsumer) 

subscribe(  @Nullable Consumer<? super T> consumer,
            @Nullable Consumer<? super Throwable> errorConsumer,
            @Nullable Runnable completeConsumer) 

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

用起來非常方便，但是響應式流規範中只定義了一個訂閱方法subscribe(Subscriber subscriber)。實際上，這幾個方法最終都是呼叫的subscribe(LambdaSubscriber subscriber)，並通過LambdaSubscriber實現了對不同個數引數的組裝。如下圖所示：

title

因此，

    flux.subscribe(System.out::println, System.err::println);

是呼叫的：

    flux.subscribe(new LambdaSubscriber(System.out::println, System.err::println, null, null));

相關推薦

（11）照虎畫貓深入理解響應式流規範——響應式Spring的道法術器

2 響應式程式設計之法 上一章本著“快速上手”的原則，介紹了響應式流的概念，以及Reactor 3的使用。這一章，我們基於Reactor 3的實現原理，從《響應式流規範》入手，深入瞭解響應式流開發庫。 2.1 響應式流規範 現代軟體對近乎實時地

（無）webservice執行過程深入理解

前面我們搞了1，2個DEMO，基本對webservice服務釋出，呼叫 ，執行 有一定的瞭解。 今天的話，我們再系統的梳理下webservice執行過程。   首先我們在webservice伺服器端開發webservice介面，然後釋出webservice；   然後我們通過一個

Android 小米推送（MiPush）的化簡深入理解

      小米推送(MiPush)是小米公司向開發者提供的訊息推送服務，總的流程如官方文件所示：      由圖可知，推送是雙向的：      1.推送是可以由app的後臺端發起，應用伺服器的後

javascript學習記錄（三）-this物件的深入理解

this 物件是基於執行的環境而決定的，this物件分為兩類（本人認為的）：全域性this以及區域性this。 全域性this我們通常是指向Windows的，區域性this是指向函式內部的。但是也有特殊的情況-閉包。如下程式碼'use strick'; var name =

【響應式程式設計的思維藝術】 （4）從打飛機遊戲理解併發與流的融合

目錄 一. 劃重點 二. 從理論到實踐 三. 問題及反思 四. 參考程式碼及Demo說明 . 本文是Rxjs 響應式程式設計-第三章: 構建併發程式這篇文章的學習筆記。 示例程式碼託管在：http://www.github.com/dashnowords/blogs

【學習筆記】深入理解js原型和閉包（11）——執行上下文棧

繼續上文的內容。 執行全域性程式碼時，會產生一個執行上下文環境，每次呼叫函式都又會產生執行上下文環境。當函式呼叫完成時，這個上下文環境以及其中的資料都會被消除，再重新回到全域性上下文環境。處於活動狀態的執行上下文環境只有一個。 其實這是一個壓棧出棧的過程——執行上下文棧。如下圖：   可

Java基礎面試題（11）----Servlet的理解和生命週期

問題 說說對servlet的理解？ 什麼是servlet的生命週期？ 回答 Servlet是什麼？ selvet（server applet），全稱Java Servlet,使用Java語言編寫的服務端程式，這些servlet都需要繼承HttpServlet這個

【後臺技術開發】Docker的映象和容器的區別（10張圖讓你深入理解Docke的r容器和映象）

這篇文章希望能夠幫助讀者深入理解Docker的命令，還有容器（container）和映象（image）之間的區別，並深入探討容器和執行中的容器之間的區別。 當我對Docker技術還是一知半解的時候，我發現理解Docker的命令非常困難。於是，我花了幾周的時間來學習Docker的工作原理，更確切地說，是關於

Lucene深入學習（11）Lucene的索引刪除

摘要： 之前的章節中已經說明了索引的新增，本節將介紹如何將特定的一個或多個文件從索引中刪除。 索引讀取 IndexReader是維護索引重要的工具類，它是一個抽象類，但是卻提供了一個靜態方法來獲取其子類的例項。 IndexReader reader =

Java記憶體區域劃分、記憶體分配原理（基於jdk1.7 源自 《深入理解java虛擬機器》）

執行時資料區域         Java虛擬機器在執行Java的過程中會把管理的記憶體劃分為若干個不同的資料區域。這些區域有各自的用途，以及建立和銷燬的時間，有的區域隨著虛擬機器程序的啟動而存在，而有的區域則依賴執行緒的啟動和結束而建立和銷燬。        Java虛擬機

效能測試必備知識（11）- 怎麼理解記憶體中的Buffer和Cache？

做效能測試的必備知識系列，可以看下面連結的文章哦 https://www.cnblogs.com/poloyy/category/1806772.html   快取 從 free 命令可以看到，快取其實就是 Buffer 和 Cache 兩部分的總和 字面意思，Buffer 是快取區，Cache

鏡像的分層結構 - 每天5分鐘玩轉容器技術（11）

數據 9.png upload 問題： 所有 rfi image tle acs Docker 支持通過擴展現有鏡像，創建新的鏡像。 實際上，Docker Hub 中 99% 的鏡像都是通過在 base 鏡像中安裝和配置需要的軟件構建出來的。比如我們現在構建一個新的鏡像，

每天一個linux命令（11）：cat命令

部分 ron linu mv命令 平臺 linux下 一個 介紹 inux 版權聲明更新：2017-05-15博主：LuckyAlan聯系：[email protected]/* */聲明：吃水不忘挖井人，轉載請註明出處！ 1 文章介紹 本文介紹了Linux下面

Python基礎（11）_python模塊之time模塊、rando模塊、hashlib、os模塊

路徑 固定 val 登錄密碼 rand getcwd ges ble sun 一、模塊 1、什麽是模塊：一個模塊就是一個包含了python定義和聲明的文件，文件名就是模塊名字加上.py的後綴 　　模塊的本質：模塊的本質是一個py文件 2、模塊分為三類：1）內置模塊；2）第三

Shiro學習（11）緩存機制

values next() app uri 如果 itl apache 依賴 nag Shiro提供了類似於spring的Cache抽象，即Shiro本身不實現Cache，但是對Cache進行了又抽象，方便更換不同的底層Cache實現。對於Cache的一些概念可以參考我的《

二叉樹（11）----求二叉樹的鏡像，遞歸和非遞歸方式

temp right 二叉樹 -a data nbsp rac art urn 1、二叉樹定義： typedef struct BTreeNodeElement_t_ { void *data; } BTreeNodeElement_t; type

python手記（11）------while循環（break 和 continue）

col except port 手記 點贊 進行 input code span 1.while 比 for 適用性更廣：滿足條件情況下一直進行 2.猜數字遊戲1.0版 import numpy as np ‘‘‘ 猜整數1.0版： 1.記錄次數

最全Pycharm教程（11）——Pycharm調試器之斷點篇

pes eas 觀察 project 項目 python pre 啟動 image 　　最全Pycharm教程（1）——定制外觀 　　最全Pycharm教程（2）——代碼風格 　　最全Pycharm教程（3）——代碼的調試、執行

Tomcat學習總結（11）——Linux下的Tomcat安全優化

app span must 以及 站點 with rip web.xml dev 1、web.xml配置及修改： 站點默認主頁： <welcome-file-list> <welcome-file>index.html</welcome-fil

Openlayers系列（一）關於地圖投影的理解

數據庫 nbsp 簡單的 是個 ace 投影 存儲 mongodb uid 近期開發以MongoDB為基礎的分布式地理數據管理平臺系統，被要求做一個簡單的demo給客戶進行演示。於是筆者便打算向數據庫中存儲一部分瓦片數據，寫一個簡單的存取服務器，使用Openlayers客戶