詳解Java中的reactive stream協議

Posted on by

背景

每個數據流都有一個生產者一個消費者。生產者負責產生數據，而消費者負責消費數據。如果是同步系統，生產一個消費一個沒什麼問題。但是如果在異步系統中，就會產生問題。

因為生產者無法感知消費者的狀態，不知道消費者到底是繁忙狀態還是空閑狀態，是否有能力去消費更多的數據。

一般來說數據隊列的長度都是有限的，即使沒有做限制，但是系統的內存也是有限的。當太多的數據沒有被消費的話，會導致內存溢出或者數據得不到即使處理的問題。

這時候就需要back-pressure瞭。

如果消息接收方消息處理不過來，則可以通知消息發送方，告知其正在承受壓力，需要降低負載。back-pressure是一種消息反饋機制，從而使系統得以優雅地響應負載， 而不是在負載下崩潰。

而reactive stream的目的就是用來管理異步服務的流數據交換，並能夠讓接收方自主決定接受數據的頻率。back-pressure就是reactive stream中不可或缺的一部分。

什麼是reactive stream

上面我們講到瞭reactive stream的作用，大傢應該對reactive stream有瞭一個基本的瞭解。這裡我們再給reactive stream做一個定義：

reactive stream就是一個異步stream處理的標準，它的特點就是非阻塞的back pressure。

reactive stream隻是一個標準，它定義瞭實現非阻塞的back pressure的最小區間的接口，方法和協議。

所以reactive stream其實有很多種實現的，不僅僅是java可以使用reactive stream，其他的編程語言也可以。

reactive stream隻是定義瞭最基本的功能，各大實現在實現瞭基本功能的同時可以自由擴展。

目前reactive stream最新的java版本是1.0.3，是在2019年8月23發佈的。它包含瞭java API，協議定義文件，測試工具集合和具體的實現例子。

深入瞭解java版本的reactive stream

在介紹java版本的reactive stream之前，我們先回顧一下reactive stream需要做哪些事情：

1.能夠處理無效數量的消息

2.消息處理是有順序的

3.可以異步的在組件之間傳遞消息

4.一定是非阻塞和backpressure的

為瞭實現這4個功能，reactive stream定義瞭4個接口，Publisher，Subscriber，Subscription，Processor。這四個接口實際上是一個觀察者模式的實現。接下來我們詳細來分析一下各個接口的作用和約定。

Publisher

先看下Publisher的定義：

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

Publisher就是用來生成消息的。它定義瞭一個subscribe方法，傳入一個Subscriber。這個方法用來將Publisher和Subscriber進行連接。

一個Publisher可以連接多個Subscriber。

每次調用subscribe建立連接，都會創建一個新的Subscription，Subscription和subscriber是一一對應的。

一個Subscriber隻能夠subscribe一次Publisher。

如果subscribe失敗或者被拒絕，則會出發Subscriber.onError(Throwable)方法。

Subscriber

先看下Subscriber的定義：

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

Subscriber就是消息的接收者。

在Publisher和Subscriber建立連接的時候會觸發onSubscribe(Subscription s)方法。

當調用Subscription.request(long)方法時，onNext(T t)會被觸發，根據request請求參數的大小，onNext會被觸發一次或者多次。

在發生異常或者結束時會觸發onError(Throwable t)或者onComplete()方法。

Subscription

先看下Subscription的定義：

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

Subscription代表著一對一的Subscriber和Publisher之間的Subscribe關系。

request(long n)意思是向publisher請求多少個events，這會觸發Subscriber.onNext方法。

cancel()則是請求Publisher停止發送信息，並清除資源。

Processor

先看下Processor的定義：

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

Processor即是Subscriber又是Publisher，它代表著一種處理狀態。

JDK中reactive stream的實現

在JDK中java.util.concurrent.Flow就是reactive stream語義的一種實現。

Flow從JDK9就開始有瞭。我們看下它的結構：

從上圖我們可以看到在JDK中Flow是一個final class,而Subscriber,Publisher,Subscription,Processor都是它的內部類。

總結

reactive stream的出現有效的解決瞭異步系統中的背壓問題。隻不過reactive stream隻是一個接口標準或者說是一種協議，具體的實現還需要自己去實現。

以上就是詳解Java中的reactive stream協議的詳細內容，更多關於Java中的reactive stream協議的資料請關註WalkonNet其它相關文章！

推薦閱讀：