Java反應式框架Reactor中的Mono和Flux

1. 前言

最近寫關於響應式編程的東西有點多，不少同窗反映對Flux和Mono這兩個Reactor中的概念有點懵逼。可是目前Java響應式編程中咱們對這兩個對象的接觸又最多，諸如Spring WebFlux、RSocket、R2DBC。我開始也對這兩個對象頭疼，因此今天咱們就簡單來探討一下它們。

2. 響應流的特色

要搞清楚這兩個概念，必須說一下響應流規範。它是響應式編程的基石。他具備如下特色：

• 響應流必須是無阻塞的。
• 響應流必須是一個數據流。
• 它必須能夠異步執行。
• 而且它也應該可以處理背壓。

背壓是反應流中的一個重要概念，能夠理解爲，生產者能夠感覺到消費者反饋的消費壓力，並根據壓力進行動態調整生產速率。形象點能夠按照下面理解：

3. Publisher

因爲響應流的特色，咱們不能再返回一個簡單的POJO對象來表示結果了。必須返回一個相似Java中的Future的概念，在有結果可用時通知消費者進行消費響應。

Reactive Stream規範中這種被定義爲Publisher<T> ，Publisher<T>是一個能夠提供0-N個序列元素的提供者，並根據其訂閱者Subscriber<? super T>的需求推送元素。一個Publisher<T>能夠支持多個訂閱者，並能夠根據訂閱者的邏輯進行推送序列元素。下面這個Excel計算就能說明一些Publisher<T>的特色。

A1-A9就能夠看作Publisher<T>及其提供的元素序列。A10-A13分別是求和函數SUM(A1:A9)、平均函數AVERAGE(A1:A9)、最大值函數MAX(A1:A9)、最小值函數MIN(A1:A9)，能夠看做訂閱者Subscriber。假如說咱們沒有A10-A13，那麼A1-A9就沒有實際意義，它們並不產生計算。這也是響應式的一個重要特色：當沒有訂閱時發佈者什麼也不作。

而Flux和Mono都是Publisher<T>在Reactor 3實現。Publisher<T>提供了subscribe方法，容許消費者在有結果可用時進行消費。若是沒有消費者Publisher<T>不會作任何事情，他根據消費狀況進行響應。 Publisher<T>可能返回零或者多個，甚至多是無限的，爲了更加清晰表示期待的結果就引入了兩個實現模型Mono和Flux。

4. Flux

Flux 是一個發出(emit)0-N個元素組成的異步序列的Publisher<T>,能夠被onComplete信號或者onError信號所終止。在響應流規範中存在三種給下游消費者調用的方法 onNext, onComplete, 和onError。下面這張圖表示了Flux的抽象模型：

以上的的講解對於初次接觸反應式編程的依然是難以理解的，因此這裏有一個按部就班的理解過程。

有些類比並非很穩當，可是對於你按部就班的理解這些新概念仍是有幫助的。

傳統數據處理

咱們在日常是這麼寫的：

public List<ClientUser> allUsers() {
    return Arrays.asList(new ClientUser("felord.cn", "reactive"),
            new ClientUser("Felordcn", "Reactor"));
}

咱們經過迭代返回值List來get這些元素進行再處理（消費），這種方式有點相似廚師作了不少菜，吃不吃在於食客。須要食客主動去來吃就好了（pull的方式），至於喜歡吃什麼不喜歡吃什麼本身隨意，怎麼吃也本身隨意。

流式數據處理

在Java 8中咱們能夠改寫爲流的表示：

public Stream<ClientUser> allUsers() {
    return  Stream.of(new ClientUser("felord.cn", "reactive"),
            new ClientUser("Felordcn", "Reactor"));
}

依然是廚師作了不少菜，可是這種就更加高級了一些，提供了菜品的搭配方式（不包含具體細節），食客能夠按照說明根據本身的習慣搭配着去吃，一但開始概不退換，吃完爲止，過時不候。

反應式數據處理

在Reactor中咱們又能夠改寫爲Flux表示：

public Flux<ClientUser> allUsers(){
    return Flux.just(new ClientUser("felord.cn", "reactive"),
            new ClientUser("Felordcn", "Reactor"));
}

這時候食客只須要訂餐就好了，作好了天然就呈上來，並且能夠隨時根據食客的飯量進行調整。若是沒有食客訂餐那麼廚師就什麼都不用作。固然不止有這麼點特性，不過對於方便咱們理解來講這就夠了。

5. Mono

Mono 是一個發出(emit)0-1個元素的Publisher<T>,能夠被onComplete信號或者onError信號所終止。

這裏就不翻譯了，總體和Flux差很少，只不過這裏只會發出0-1個元素。也就是說不是有就是沒有。象Flux同樣，咱們來看看Mono的演化過程以幫助理解。

傳統數據處理

public ClientUser currentUser () {
    return isAuthenticated ? new ClientUser("felord.cn", "reactive") : null;
}

直接返回符合條件的對象或者null。

Optional的處理方式

public Optional<ClientUser> currentUser () {
    return isAuthenticated ? Optional.of(new ClientUser("felord.cn", "reactive"))
            : Optional.empty();
}

這個Optional我以爲就有反應式的那種味兒了，固然它並非反應式。當咱們不從返回值Optional取其中具體的對象時，咱們不清楚裏面到底有沒有，可是Optional是必定客觀存在的,不會出現NPE問題。

反應式數據處理

public Mono<ClientUser> currentUser () {
    return isAuthenticated ? Mono.just(new ClientUser("felord.cn", "reactive"))
            : Mono.empty();
}

和Optional有點相似的機制，固然Mono不是爲了解決NPE問題的，它是爲了處理響應流中單個值（也多是Void）而存在的。

6. 總結

Flux和Mono是Java反應式中的重要概念，可是不少同窗包括我在開始都難以理解它們。這實際上是規定了兩種流式範式，這種範式讓數據具備一些新的特性，好比基於發佈訂閱的事件驅動，異步流、背壓等等。另外數據是推送（Push）給消費者的以區別於平時咱們的拉（Pull）模式。同時咱們能夠像Stream Api同樣使用相似map、flatmap等操做符（operator）來操做它們。對Flux和Mono這兩個概念須要花一些時間去理解它們，不能操之過急。若是你對個人這種見解有不一樣的觀點能夠留言討論，多多關注：碼農小胖哥 獲取更多幹貨知識。

關注公衆號：Felordcn 獲取更多資訊

我的博客：https://felord.cn