背景
Stream大家應該都很熟悉了，java8中為所有的集合類都引入了Stream的概念。優雅的鏈式操作，流式處理邏輯，相信用過的人都會愛不釋手。
每個數據流都有一個生產者一個消費者。生產者負責產生數據，而消費者負責消費數據。如果是同步系統，生產一個消費一個沒什么問題。但是如果在異步系統中，就會產生問題。
因為生產者無法感知消費者的狀態，不知道消費者到底是繁忙狀態還是空閑狀態，是否有能力去消費更多的數據。
一般來說數據隊列的長度都是有限的，即使沒有做限制，但是系統的內存也是有限的。當太多的數據沒有被消費的話，會導致內存溢出或者數據得不到即使處理的問題。
這時候就需要back-pressure了。
如果消息接收方消息處理不過來，則可以通知消息發送方，告知其正在承受壓力，需要降低負載。back-pressure是一種消息反饋機制，從而使系統得以優雅地響應負載， 而不是在負載下崩潰。
而reactive stream的目的就是用來管理異步服務的流數據交換，並能夠讓接收方自主決定接受數據的頻率。back-pressure就是reactive stream中不可或缺的一部分。
更多內容請訪問www.flydean.com
什么是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需要做哪些事情：
- 能夠處理無效數量的消息
- 消息處理是有順序的
- 可以異步的在組件之間傳遞消息
- 一定是非阻塞和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都是它的內部類。
我們會在后面的文章中繼續講解JDK中Flow的使用。敬請期待。
總結
reactive stream的出現有效的解決了異步系統中的背壓問題。只不過reactive stream只是一個接口標准或者說是一種協議，具體的實現還需要自己去實現。
本文作者：flydean程序那些事
本文鏈接：http://www.flydean.com/reactive-stream-protocol/
本文來源：flydean的博客
歡迎關注我的公眾號:程序那些事，更多精彩等着您！