[[402929]]

 1.Backpressure

Backpressure 在國內(nèi)被翻譯成背壓，這個翻譯在網(wǎng)上被很多人吐槽，我覺得大家的吐槽是有道理的，背壓單純從字面上確實看不出來有什么意思。所以松哥這里直接用英文 Backpressure 吧。

Backpressure 是一種現(xiàn)象：當(dāng)數(shù)據(jù)流從上游生產(chǎn)者向下游消費者傳輸?shù)倪^程中，上游生產(chǎn)速度大于下游消費速度，導(dǎo)致下游的 Buffer 溢出，這種現(xiàn)象就叫做 Backpressure。

換句話說，上游生產(chǎn)數(shù)據(jù)，生產(chǎn)完成后通過管道將數(shù)據(jù)傳到下游，下游消費數(shù)據(jù)，當(dāng)下游消費速度小于上游數(shù)據(jù)生產(chǎn)速度時，數(shù)據(jù)在管道中積壓會對上游形成一個壓力，這就是 Backpressure，從這個角度來說，Backpressure 翻譯成反壓、回壓似乎更合理一些。

Backpressure 會出現(xiàn)在有 Buffer 上限的系統(tǒng)中，當(dāng)出現(xiàn) Buffer 溢出的時候，就會有 Backpressure，對于 Backpressure，它的應(yīng)對措施只有一個：丟棄新事件。那么什么是 Buffer 溢出呢?例如我的服務(wù)器可以同時處理 2000 個用戶請求，那么我就把請求上限設(shè)置為 2000，這個 2000 就是我的 Buffer，當(dāng)超出 2000 的時候，就產(chǎn)生了 Backpressure。

2.Flow API

JDK9 中推出了 Flow API，用以支持 Reactive Programming，即響應(yīng)式編程。

在響應(yīng)式編程中，會有一個數(shù)據(jù)發(fā)布者 Publisher 和數(shù)據(jù)訂閱者 Subscriber，Subscriber 接收 Publisher 發(fā)布的數(shù)據(jù)并進(jìn)行消費，在 Subscriber 和 Publisher 之間還存在一個 Processor，類似于一個過濾器，可以對數(shù)據(jù)進(jìn)行中間處理。

JDK9 中提供了 Flow API 用以支持響應(yīng)式編程，另外 RxJava 和 Reactor 等框架也提供了相關(guān)的實現(xiàn)。

我們來看看 JDK9 中的 Flow 類：

非常簡潔，基本上就是按照 Reactive Programming 的設(shè)計來的：

Publisher

Publisher 為數(shù)據(jù)發(fā)布者，這是一個函數(shù)式接口，里邊只有一個方法，通過這個方法將數(shù)據(jù)發(fā)布出去，Publisher 的定義如下：

@FunctionalInterface 
public static interface Publisher<T> { 
    public void subscribe(Subscriber<? super T> subscriber); 
} 

Subscriber

Subscriber 為數(shù)據(jù)訂閱者，這個里邊有四個方法，如下：

public static interface Subscriber<T> { 
    public void onSubscribe(Subscription subscription); 
    public void onNext(T item); 
    public void onError(Throwable throwable); 
    public void onComplete(); 
} 

• onSubscribe：這個是訂閱成功的回調(diào)方法，用于初始化 Subscription，并且表明可以開始接收訂閱數(shù)據(jù)了。
• onNext：接收下一項訂閱數(shù)據(jù)的回調(diào)方法。
• onError：在 Publisher 或 Subcriber 遇到不可恢復(fù)的錯誤時調(diào)用此方法，之后 Subscription 不會再調(diào)用 Subscriber 其他的方法。
• onComplete：當(dāng)接收完所有訂閱數(shù)據(jù)，并且發(fā)布者已經(jīng)關(guān)閉后會回調(diào)這個方法。

Subscription

Subscription 為發(fā)布者和訂閱者之間的訂閱關(guān)系，用來控制消息的消費，這個里邊有兩個方法：

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

• request：這個方法用來向數(shù)據(jù)發(fā)布者請求 n 個數(shù)據(jù)。
• cancel：取消消息訂閱，訂閱者將不再接收數(shù)據(jù)。

Processor

Processor 是一個空接口，不過它同時繼承了 Publisher 和 Subscriber，所以它既能發(fā)布數(shù)據(jù)也能訂閱數(shù)據(jù)，因此我們可以通過 Processor 來完成一些數(shù)據(jù)轉(zhuǎn)換的功能，先接收數(shù)據(jù)進(jìn)行處理，處理完成后再將數(shù)據(jù)發(fā)布出去，這個也有點類似于我們 JavaEE 中的過濾器。

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

2.1 消息訂閱初體驗

我們通過如下一段代碼體驗一下消息的訂閱與發(fā)布：

public class FlowDemo { 
    public static void main(String[] args) { 
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>(); 
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() { 
            private Flow.Subscription subscription; 
            @Override 
            public void onSubscribe(Flow.Subscription subscription) { 
                this.subscription = subscription; 
                //向數(shù)據(jù)發(fā)布者請求一個數(shù)據(jù) 
                this.subscription.request(1); 
            } 
            @Override 
            public void onNext(String item) { 
                System.out.println("接收到 publisher 發(fā)來的消息了：" + item); 
                //接收完成后，可以繼續(xù)接收或者不接收 
                //this.subscription.cancel(); 
                this.subscription.request(1); 
            } 
            @Override 
            public void onError(Throwable throwable) { 
                //出現(xiàn)異常，就會來到這個方法，此時直接取消訂閱即可 
                this.subscription.cancel(); 
            } 
            @Override 
            public void onComplete() { 
                //發(fā)布者的所有數(shù)據(jù)都被接收，并且發(fā)布者已經(jīng)關(guān)閉 
                System.out.println("數(shù)據(jù)接收完畢"); 
            } 
        }; 
        //配置發(fā)布者和訂閱者 
        publisher.subscribe(subscriber); 
        for (int i = 0; i < 5; i++) { 
            //發(fā)送數(shù)據(jù) 
            publisher.submit("hello:" + i); 
        } 
        //關(guān)閉發(fā)布者 
        publisher.close(); 
        new Scanner(System.in).next(); 
    } 
} 

松哥稍微解釋一下上面這段代碼：

1. 首先創(chuàng)建一個 SubmissionPublisher 對象作為消息發(fā)布者。
2. 接下來創(chuàng)建 Flow.Subscriber 對象作為消息訂閱者，實現(xiàn)消息訂閱者里邊的四個方法，分別進(jìn)行處理。
3. 為 publisher 配置上 subscriber。
4. 發(fā)送消息。
5. 消息發(fā)送完成后關(guān)閉 publisher。
6. 最后是讓程序不要停止，觀察消息訂閱者打印情況。

2.2 模擬 Backpressure

Backpressure 問題在 Flow API 中得到了很好的解決。Subscriber 會將 Publisher 發(fā)布的數(shù)據(jù)緩存在 Subscription 中，其長度默認(rèn)為256，相關(guān)源碼如下：

public final class Flow { 
    static final int DEFAULT_BUFFER_SIZE = 256; 
    public static int defaultBufferSize() { 
        return DEFAULT_BUFFER_SIZE; 
    } 
    ... 
} 

一旦超出這個數(shù)據(jù)量，publisher 就會降低數(shù)據(jù)發(fā)送速度。

我們對上面的案例進(jìn)行修改，如下：

public class FlowDemo { 
    public static void main(String[] args) { 
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>(); 
 
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() { 
            private Flow.Subscription subscription; 
 
            @Override 
            public void onSubscribe(Flow.Subscription subscription) { 
                this.subscription = subscription; 
                //向數(shù)據(jù)發(fā)布者請求一個數(shù)據(jù) 
                this.subscription.request(1); 
            } 
 
            @Override 
            public void onNext(String item) { 
                System.out.println("接收到 publisher 發(fā)來的消息了：" + item); 
                //接收完成后，可以繼續(xù)接收或者不接收 
                //this.subscription.cancel(); 
                try { 
                    Thread.sleep(2000); 
                } catch (InterruptedException e) { 
                    e.printStackTrace(); 
                } 
                this.subscription.request(1); 
            } 
 
            @Override 
            public void onError(Throwable throwable) { 
                //出現(xiàn)異常，就會來到這個方法，此時直接取消訂閱即可 
                this.subscription.cancel(); 
            } 
 
            @Override 
            public void onComplete() { 
                //發(fā)布者的所有數(shù)據(jù)都被接收，并且發(fā)布者已經(jīng)關(guān)閉 
                System.out.println("數(shù)據(jù)接收完畢"); 
            } 
        }; 
        publisher.subscribe(subscriber); 
        for (int i = 0; i < 500; i++) { 
            System.out.println("i--------->" + i); 
            publisher.submit("hello:" + i); 
        } 
        //關(guān)閉發(fā)布者 
        publisher.close(); 
        new Scanner(System.in).next(); 
    } 
} 

一共修改了三個地方：

1. Subscriber#onNext 方法中，每次休息兩秒再處理下一條數(shù)據(jù)。
2. 發(fā)布數(shù)據(jù)時，一共發(fā)布 500 條數(shù)據(jù)。
3. 打印數(shù)據(jù)發(fā)布的日志。

修改完成后，我們再次啟動項目，觀察控制臺輸出：

可以看到，生產(chǎn)者先是一股腦生產(chǎn)了 257 條數(shù)據(jù)(hello0 在一開始就被消費了，所以緩存中實際上是 256 條)，消息則是一條一條的來，由于消費的速度比較慢，所以當(dāng)緩存中的數(shù)據(jù)超過 256 條之后，接下來都是消費一條，再發(fā)送一條。

2.3 數(shù)據(jù)處理

Flow.Processor 可以像過濾器一樣，對數(shù)據(jù)進(jìn)行預(yù)處理，數(shù)據(jù)從 publisher 出來之后，先進(jìn)入 Flow.Processor 中進(jìn)行預(yù)處理，然后再進(jìn)入 Subscriber。

修改后的代碼如下：

public class FlowDemo { 
    public static void main(String[] args) { 
 
        class DataFilter extends SubmissionPublisher<String> implements Flow.Processor<String,String>{ 
 
            private Flow.Subscription subscription; 
 
            @Override 
            public void onSubscribe(Flow.Subscription subscription) { 
                this.subscription = subscription; 
                this.subscription.request(1); 
            } 
 
            @Override 
            public void onNext(String item) { 
                this.submit("【這是一條被處理過的數(shù)據(jù)】" + item); 
                this.subscription.request(1); 
            } 
 
            @Override 
            public void onError(Throwable throwable) { 
                this.subscription.cancel(); 
            } 
 
            @Override 
            public void onComplete() { 
                this.close(); 
            } 
        } 
 
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>(); 
        DataFilter dataFilter = new DataFilter(); 
        publisher.subscribe(dataFilter); 
 
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() { 
            private Flow.Subscription subscription; 
 
            @Override 
            public void onSubscribe(Flow.Subscription subscription) { 
                this.subscription = subscription; 
                //向數(shù)據(jù)發(fā)布者請求一個數(shù)據(jù) 
                this.subscription.request(1); 
            } 
 
            @Override 
            public void onNext(String item) { 
                System.out.println("接收到 publisher 發(fā)來的消息了：" + item); 
                //接收完成后，可以繼續(xù)接收或者不接收 
                //this.subscription.cancel(); 
                try { 
                    Thread.sleep(2000); 
                } catch (InterruptedException e) { 
                    e.printStackTrace(); 
                } 
                this.subscription.request(1); 
            } 
 
            @Override 
            public void onError(Throwable throwable) { 
                //出現(xiàn)異常，就會來到這個方法，此時直接取消訂閱即可 
                this.subscription.cancel(); 
            } 
 
            @Override 
            public void onComplete() { 
                //發(fā)布者的所有數(shù)據(jù)都被接收，并且發(fā)布者已經(jīng)關(guān)閉 
                System.out.println("數(shù)據(jù)接收完畢"); 
            } 
        }; 
        dataFilter.subscribe(subscriber); 
        for (int i = 0; i < 500; i++) { 
            System.out.println("發(fā)送消息 i--------->" + i); 
            publisher.submit("hello:" + i); 
        } 
        //關(guān)閉發(fā)布者 
        publisher.close(); 
        new Scanner(System.in).next(); 
    } 
} 

簡單起見，我這里創(chuàng)建了一個局部內(nèi)部類 DataFilter，DataFilter 繼承自 SubmissionPublisher 并實現(xiàn)了 Flow.Processor 接口，由于 DataFilter 繼承自 SubmissionPublisher，所以它也兼具 SubmissionPublisher 的功能。

在 DataFilter 中完成消息的處理并重新發(fā)送出去。接下來定義 publisher，讓 dataFilter 作為其訂閱者，再定義新的訂閱者，作為 dataFilter 的訂閱者。

最終運行效果如下：

3.小結(jié)

好啦，這就是今天和大家介紹的 Java9 中的 Reactive Stream，那么至此，我們的 WebFlux 前置知識差不多告一段落了，下篇文章開始，正式開整 WebFlux。

本文轉(zhuǎn)載自微信公眾號「江南一點雨」，可以通過以下二維碼關(guān)注。轉(zhuǎn)載本文請聯(lián)系江南一點雨公眾號。