file

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

Java 9的 Reactive Streams是對異步流式編程的一種實(shí)現(xiàn)。它基于異步發(fā)布和訂閱模型，具有非阻塞“背壓”數(shù)據(jù)處理的特點(diǎn)。

Non-blocking Back Pressure(非阻塞背壓)：它是一種機(jī)制，讓發(fā)布訂閱模型中的訂閱者避免接收大量數(shù)據(jù)(超出其處理能力)，訂閱者可以異步通知發(fā)布者降低或提升數(shù)據(jù)生產(chǎn)發(fā)布的速率。它是響應(yīng)式編程實(shí)現(xiàn)效果的核心特點(diǎn)！

一、Java9 Reactive Stream API

Java 9提供了一組定義響應(yīng)式流編程的接口。所有這些接口都作為靜態(tài)內(nèi)部接口定義在java.util.concurrent.Flow類里面。

file

下面是Java 響應(yīng)式編程中的一些重要角色和概念，先簡單理解一下

發(fā)布者(Publisher)是潛在的無限數(shù)量的有序數(shù)據(jù)元素的生產(chǎn)者。 它根據(jù)收到的需求(subscription)向當(dāng)前訂閱者發(fā)布一定數(shù)量的數(shù)據(jù)元素。

訂閱者(Subscriber)從發(fā)布者那里訂閱并接收數(shù)據(jù)元素。與發(fā)布者建立訂閱關(guān)系后，發(fā)布者向訂閱者發(fā)送訂閱令牌(subscription)，訂閱者可以根據(jù)自己的處理能力請求發(fā)布者發(fā)布數(shù)據(jù)元素的數(shù)量。

訂閱令牌(subscription)表示訂閱者與發(fā)布者之間建立的訂閱關(guān)系。 當(dāng)建立訂閱關(guān)系后，發(fā)布者將其傳遞給訂閱者。 訂閱者使用訂閱令牌與發(fā)布者進(jìn)行交互，例如請求數(shù)據(jù)元素的數(shù)量或取消訂閱。

二、Java響應(yīng)式編程四大接口

2.1.Subscriber Interface(訂閱者訂閱接口)

public static interface Subscriber {

public void onSubscribe(Subscription subscription);

public void onNext(T item);

public void onError(Throwable throwable);

public void onComplete();

}

onSubscribe：在發(fā)布者接受訂閱者的訂閱動(dòng)作之后，發(fā)布任何的訂閱消息之前被調(diào)用。新創(chuàng)建的Subscription訂閱令牌對象通過此方法傳遞給訂閱者。

onNext：下一個(gè)待處理的數(shù)據(jù)項(xiàng)的處理函數(shù)

onError：在發(fā)布者或訂閱遇到不可恢復(fù)的錯(cuò)誤時(shí)調(diào)用

onComplete：當(dāng)沒有訂閱者調(diào)用(包括onNext()方法)發(fā)生時(shí)調(diào)用。

2.2.Subscription Interface (訂閱令牌接口)

訂閱令牌對象通過Subscriber.onSubscribe()方法傳遞

public static interface Subscription {

public void request(long n);

public void cancel();

}

request(long n)是無阻塞背壓概念背后的關(guān)鍵方法。訂閱者使用它來請求n個(gè)以上的消費(fèi)項(xiàng)目。這樣，訂閱者控制了它當(dāng)前能夠接收多少個(gè)數(shù)據(jù)。

cancel()由訂閱者主動(dòng)來取消其訂閱，取消后將不會(huì)在接收到任何數(shù)據(jù)消息。

2.3.Publisher Interface(發(fā)布者接口)

@FunctionalInterface

public static interface Publisher {

public void subscribe(Subscriber super T> subscriber);

}

調(diào)用該方法，建立訂閱者Subscriber與發(fā)布者Publisher之間的消息訂閱關(guān)系。

2.4.Processor Interface(處理器接口)

處理者Processor 可以同時(shí)充當(dāng)訂閱者和發(fā)布者，起到轉(zhuǎn)換發(fā)布者——訂閱者管道中的元素的作用。用于將發(fā)布者T類型的數(shù)據(jù)元素，接收并轉(zhuǎn)換為類型R的數(shù)據(jù)并發(fā)布。

public static interface Processor extends Subscriber, Publisher {

}

二、實(shí)戰(zhàn)案例

現(xiàn)在我們要去實(shí)現(xiàn)上面的四個(gè)接口來完成響應(yīng)式編程

Subscription Interface訂閱令牌接口通常不需要我們自己編程去實(shí)現(xiàn)，我們只需要在知道request()方法和cancle()方法含義即可。

Publisher Interface發(fā)布者接口，Java 9 已經(jīng)默認(rèn)為我們提供了實(shí)現(xiàn)SubmissionPublisher，該實(shí)現(xiàn)類除了實(shí)現(xiàn)Publisher接口的方法外，提供了一個(gè)方法叫做submit()來完成消息數(shù)據(jù)的發(fā)送。

Subscriber Interface訂閱者接口，通常需要我們自己去實(shí)現(xiàn)。因?yàn)樵跀?shù)據(jù)訂閱接收之后，不同的業(yè)務(wù)有不同的處理邏輯。

Processor實(shí)際上是 Publisher Interface和Subscriber Interface的集合體，有需要數(shù)據(jù)類型轉(zhuǎn)換及數(shù)據(jù)處理的需求才去實(shí)現(xiàn)這個(gè)接口

下面的例子實(shí)現(xiàn)的式字符串的數(shù)據(jù)消息訂閱處理

實(shí)現(xiàn)訂閱者Subscriber Interface

import java.util.concurrent.Flow;

public class MySubscriber implements Flow.Subscriber {

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

@Override

public void onSubscribe(Flow.Subscription subscription) {

System.out.println("訂閱關(guān)系建立onSubscribe: " + subscription);

this.subscription = subscription;

subscription.request(2);

}

@Override

public void onNext(String item) {

System.out.println("item: " + item);

// 一個(gè)消息處理完成之后，可以繼續(xù)調(diào)用subscription.request(n);向發(fā)布者要求數(shù)據(jù)發(fā)送

//subscription.request(n);

}

@Override

public void onError(Throwable throwable) {

System.out.println("onError: " + throwable);

}

@Override

public void onComplete() {

System.out.println("onComplete");

}

}

SubmissionPublisher消息發(fā)布者

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 sb = new SubmissionPublisher<>(executor, Flow.defaultBufferSize());

sb.subscribe(new MySubscriber()); //建立訂閱關(guān)系，可以有多個(gè)訂閱者

sb.submit("數(shù)據(jù) 1"); //發(fā)送消息1

sb.submit("數(shù)據(jù) 2"); //發(fā)送消息2

sb.submit("數(shù)據(jù) 3"); //發(fā)送消息3

executor.shutdown();

}

}

控制臺(tái)打印輸出結(jié)果

訂閱關(guān)系建立

onSubscribe: java.util.concurrent.SubmissionPublisher$BufferedSubscription@27e81a39

item: 數(shù)據(jù) 1

item: 數(shù)據(jù) 2

請注意：即使發(fā)布者submit了3條數(shù)據(jù)，MySubscriber也僅收到了2條數(shù)據(jù)進(jìn)行了處理。是因?yàn)槲覀冊贛ySubscriber#onSubscribe()方法中使用了subscription.request(2);。這就是“背壓”的響應(yīng)式編程效果，我有能力處理多少數(shù)據(jù)，就會(huì)通知消息發(fā)布者給多少數(shù)據(jù)。

歡迎關(guān)注我的博客，里面有很多精品合集

本文轉(zhuǎn)載注明出處(必須帶連接，不能只轉(zhuǎn)文字)：字母哥博客。

覺得對您有幫助的話，幫我點(diǎn)贊、分享！您的支持是我不竭的創(chuàng)作動(dòng)力！ 。另外，筆者最近一段時(shí)間輸出了如下的精品內(nèi)容，期待您的關(guān)注。

總結(jié)

以上是生活随笔為你收集整理的java9 反应编程_Java9第四篇-Reactive Stream API响应式编程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。