Java9第四篇-Reactive Stream API回應式編程-有解無憂

file

我計劃在后續的一段時間內，寫一系列關于java 9的文章，雖然java 9 不像Java 8或者Java 11那樣的核心java版本，但是還是有很多的特性值得關注，期待您能關注我，我將把java 9 寫成一系列的文章，大概十篇左右，

java9第一篇-可以在interface中定義私有方法了
java9第二篇-Java9改進try-with-resources語法
java9第三篇-支持多JDK版本下運行的Jar檔案打包方式

Java 9的 Reactive Streams是對異步流式編程的一種實作，它基于異步發布和訂閱模型，具有非阻塞“背壓”資料處理的特點，

Non-blocking Back Pressure(非阻塞背壓)：它是一種機制，讓發布訂閱模型中的訂閱者避免接收大量資料（超出其處理能力），訂閱者可以異步通知發布者降低或提升資料生產發布的速率，它是回應式編程實作效果的核心特點！

一、Java9 Reactive Stream API

Java 9提供了一組定義回應式流編程的介面，所有這些介面都作為靜態內部介面定義在java.util.concurrent.Flow類里面，

file

下面是Java 回應式編程中的一些重要角色和概念，先簡單理解一下

發布者（Publisher）是潛在的無限數量的有序資料元素的生產者，它根據收到的需求(subscription)向當前訂閱者發布一定數量的資料元素，
訂閱者（Subscriber）從發布者那里訂閱并接收資料元素，與發布者建立訂閱關系后，發布者向訂閱者發送訂閱令牌（subscription），訂閱者可以根據自己的處理能力請求發布者發布資料元素的數量，
訂閱令牌（subscription）表示訂閱者與發布者之間建立的訂閱關系，當建立訂閱關系后，發布者將其傳遞給訂閱者，訂閱者使用訂閱令牌與發布者進行互動，例如請求資料元素的數量或取消訂閱，

二、Java回應式編程四大介面

2.1.Subscriber Interface（訂閱者訂閱介面）

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

onSubscribe：在發布者接受訂閱者的訂閱動作之后，發布任何的訂閱訊息之前被呼叫，新創建的Subscription訂閱令牌物件通過此方法傳遞給訂閱者，
onNext：下一個待處理的資料項的處理函式
onError：在發布者或訂閱遇到不可恢復的錯誤時呼叫
onComplete：當沒有訂閱者呼叫（包括onNext()方法）發生時呼叫，

2.2.Subscription Interface （訂閱令牌介面）

訂閱令牌物件通過Subscriber.onSubscribe()方法傳遞

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

request(long n)是無阻塞背壓概念背后的關鍵方法，訂閱者使用它來請求n個以上的消費專案，這樣，訂閱者控制了它當前能夠接收多少個資料，
cancel()由訂閱者主動來取消其訂閱，取消后將不會在接收到任何資料訊息，

2.3.Publisher Interface(發布者介面)

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

呼叫該方法，建立訂閱者Subscriber與發布者Publisher之間的訊息訂閱關系，

2.4.Processor Interface（處理器介面）

處理者Processor 可以同時充當訂閱者和發布者，起到轉換發布者——訂閱者管道中的元素的作用，用于將發布者T型別的資料元素，接收并轉換為型別R的資料并發布，

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

二、實戰案例

現在我們要去實作上面的四個介面來完成回應式編程

Subscription Interface訂閱令牌介面通常不需要我們自己編程去實作，我們只需要在知道request()方法和cancle()方法含義即可，
Publisher Interface發布者介面，Java 9 已經默認為我們提供了實作SubmissionPublisher，該實作類除了實作Publisher介面的方法外，提供了一個方法叫做submit()來完成訊息資料的發送，
Subscriber Interface訂閱者介面，通常需要我們自己去實作，因為在資料訂閱接收之后，不同的業務有不同的處理邏輯，
Processor實際上是 Publisher Interface和Subscriber Interface的集合體，有需要資料型別轉換及資料處理的需求才去實作這個介面

下面的例子實作的式字串的資料訊息訂閱處理

實作訂閱者Subscriber Interface

import java.util.concurrent.Flow;

public class MySubscriber implements Flow.Subscriber<String> {

  private Flow.Subscription subscription;  //訂閱令牌

  @Override
  public void onSubscribe(Flow.Subscription subscription) {
      System.out.println("訂閱關系建立onSubscribe: " + subscription);
      this.subscription = subscription;
      subscription.request(2);
  }

  @Override
  public void onNext(String item) {
      System.out.println("item: " + item);
      // 一個訊息處理完成之后，可以繼續呼叫subscription.request(n);向發布者要求資料發送
      //subscription.request(n);
  }

  @Override
  public void one rror(Throwable throwable) {
      System.out.println("onError: " + throwable);
  }

  @Override
  public void onComplete() {
      System.out.println("onComplete");
  }
}

SubmissionPublisher訊息發布者

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class SubmissionPublisherExample {
  public static void main(String[] args) throws InterruptedException {
      ExecutorService executor = Executors.newFixedThreadPool(1);
      SubmissionPublisher<String> sb = new SubmissionPublisher<>(executor, Flow.defaultBufferSize());
      sb.subscribe(new MySubscriber());   //建立訂閱關系，可以有多個訂閱者
      sb.submit("資料 1");  //發送訊息1
      sb.submit("資料 2"); //發送訊息2
      sb.submit("資料 3"); //發送訊息3

      executor.shutdown();
  }
}

控制臺列印輸出結果

訂閱關系建立
onSubscribe: java.util.concurrent.SubmissionPublisher$BufferedSubscription@27e81a39
item: 資料 1
item: 資料 2

請注意：即使發布者submit了3條資料，MySubscriber也僅收到了2條資料進行了處理，是因為我們在MySubscriber#onSubscribe()方法中使用了subscription.request(2);，這就是“背壓”的回應式編程效果，我有能力處理多少資料，就會通知訊息發布者給多少資料，

歡迎關注我的博客，里面有很多精品合集

本文轉載注明出處（必須帶連接，不能只轉文字）：字母哥博客，

覺得對您有幫助的話，幫我點贊、分享！您的支持是我不竭的創作動力！ ，另外，筆者最近一段時間輸出了如下的精品內容，期待您的關注，

《手摸手教你學Spring Boot2.0》
《Spring Security-JWT-OAuth2一本通》
《實戰前后端分離RBAC權限管理系統》
《實戰SpringCloud微服務從青銅到王者》
《VUE深入淺出系列》

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/175973.html

標籤：Java

上一篇：學習第54天

下一篇：10年經驗17張圖帶你進入gitflow企業專案代碼版本管理的最佳實踐