怎么在Java中使用reactive stream协议

这篇文章将为大家详细讲解有关怎么在Java中使用reactive stream协议，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

什么是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需要做哪些事情：

1.能够处理无效数量的消息

2.消息处理是有顺序的

3.可以异步的在组件之间传递消息

4.一定是非阻塞和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就开始有了。我们看下它的结构：

关于怎么在Java中使用reactive stream协议就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。