背景

每個(gè)數(shù)據(jù)流都有一個(gè)生產(chǎn)者一個(gè)消費(fèi)者。生產(chǎn)者負(fù)責(zé)產(chǎn)生數(shù)據(jù)，而消費(fèi)者負(fù)責(zé)消費(fèi)數(shù)據(jù)。如果是同步系統(tǒng)，生產(chǎn)一個(gè)消費(fèi)一個(gè)沒什么問題。但是如果在異步系統(tǒng)中，就會(huì)產(chǎn)生問題。

因?yàn)樯a(chǎn)者無法感知消費(fèi)者的狀態(tài)，不知道消費(fèi)者到底是繁忙狀態(tài)還是空閑狀態(tài)，是否有能力去消費(fèi)更多的數(shù)據(jù)。

一般來說數(shù)據(jù)隊(duì)列的長(zhǎng)度都是有限的，即使沒有做限制，但是系統(tǒng)的內(nèi)存也是有限的。當(dāng)太多的數(shù)據(jù)沒有被消費(fèi)的話，會(huì)導(dǎo)致內(nèi)存溢出或者數(shù)據(jù)得不到即使處理的問題。

這時(shí)候就需要back-pressure了。

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

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

什么是reactive stream

上面我們講到了reactive stream的作用，大家應(yīng)該對(duì)reactive stream有了一個(gè)基本的了解。這里我們?cè)俳oreactive stream做一個(gè)定義：

reactive stream就是一個(gè)異步stream處理的標(biāo)準(zhǔn)，它的特點(diǎn)就是非阻塞的back pressure。

reactive stream只是一個(gè)標(biāo)準(zhǔn)，它定義了實(shí)現(xiàn)非阻塞的back pressure的最小區(qū)間的接口，方法和協(xié)議。

所以reactive stream其實(shí)有很多種實(shí)現(xiàn)的，不僅僅是java可以使用reactive stream，其他的編程語言也可以。

reactive stream只是定義了最基本的功能，各大實(shí)現(xiàn)在實(shí)現(xiàn)了基本功能的同時(shí)可以自由擴(kuò)展。

目前reactive stream最新的java版本是1.0.3，是在2019年8月23發(fā)布的。它包含了java API，協(xié)議定義文件，測(cè)試工具集合和具體的實(shí)現(xiàn)例子。

深入了解java版本的reactive stream

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

1.能夠處理無效數(shù)量的消息

2.消息處理是有順序的

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

4.一定是非阻塞和backpressure的

為了實(shí)現(xiàn)這4個(gè)功能，reactive stream定義了4個(gè)接口，Publisher，Subscriber，Subscription，Processor。這四個(gè)接口實(shí)際上是一個(gè)觀察者模式的實(shí)現(xiàn)。接下來我們?cè)敿?xì)來分析一下各個(gè)接口的作用和約定。

Publisher

先看下Publisher的定義：

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

Publisher就是用來生成消息的。它定義了一個(gè)subscribe方法，傳入一個(gè)Subscriber。這個(gè)方法用來將Publisher和Subscriber進(jìn)行連接。

一個(gè)Publisher可以連接多個(gè)Subscriber。

每次調(diào)用subscribe建立連接，都會(huì)創(chuàng)建一個(gè)新的Subscription，Subscription和subscriber是一一對(duì)應(yīng)的。

一個(gè)Subscriber只能夠subscribe一次Publisher。

如果subscribe失敗或者被拒絕，則會(huì)出發(fā)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建立連接的時(shí)候會(huì)觸發(fā)onSubscribe(Subscription s)方法。

當(dāng)調(diào)用Subscription.request(long)方法時(shí)，onNext(T t)會(huì)被觸發(fā)，根據(jù)request請(qǐng)求參數(shù)的大小，onNext會(huì)被觸發(fā)一次或者多次。

在發(fā)生異?；蛘呓Y(jié)束時(shí)會(huì)觸發(fā)onError(Throwable t)或者onComplete()方法。

Subscription

先看下Subscription的定義：

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

Subscription代表著一對(duì)一的Subscriber和Publisher之間的Subscribe關(guān)系。

request(long n)意思是向publisher請(qǐng)求多少個(gè)events，這會(huì)觸發(fā)Subscriber.onNext方法。

cancel()則是請(qǐng)求Publisher停止發(fā)送信息，并清除資源。

Processor

先看下Processor的定義：

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

Processor即是Subscriber又是Publisher，它代表著一種處理狀態(tài)。

JDK中reactive stream的實(shí)現(xiàn)

在JDK中java.util.concurrent.Flow就是reactive stream語義的一種實(shí)現(xiàn)。

Flow從JDK9就開始有了。我們看下它的結(jié)構(gòu)：

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

總結(jié)

reactive stream的出現(xiàn)有效的解決了異步系統(tǒng)中的背壓?jiǎn)栴}。只不過reactive stream只是一個(gè)接口標(biāo)準(zhǔn)或者說是一種協(xié)議，具體的實(shí)現(xiàn)還需要自己去實(shí)現(xiàn)。

以上就是詳解Java中的reactive stream協(xié)議的詳細(xì)內(nèi)容，更多關(guān)于Java中的reactive stream協(xié)議的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論