文章目錄

• 背景
• 什么是reactive stream
• 深入了解java版本的reactive stream
• • Publisher
  • Subscriber
  • Subscription
  • Processor
• JDK中reactive stream的實現
• 總結

背景

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只是一個接口標準或者說是一種協議，具體的實現還需要自己去實現。

更多精彩內容且看：

• 區塊鏈從入門到放棄系列教程-涵蓋密碼學,超級賬本,以太坊,Libra,比特幣等持續更新
• Spring Boot 2.X系列教程:七天從無到有掌握Spring Boot-持續更新
• Spring 5.X系列教程:滿足你對Spring5的一切想象-持續更新
• java程序員從小工到專家成神之路（2020版）-持續更新中,附詳細文章教程

本文作者：flydean程序那些事

本文鏈接：http://www.flydean.com/reactive-stream-protocol/

本文來源：flydean的博客

歡迎關注我的公眾號:程序那些事，更多精彩等著您！

總結

以上是生活随笔為你收集整理的reactive stream协议详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。