JDK9 响应式流使用详解

2022 年 2 月 03 日
本文字数：5854 字
阅读完需：约 19 分钟

上文中咱们简单提到了 JDK9 中 Flow 接口中的静态内部类实现了响应式流的 JAVA API，并且提供了一个一个 Publisher 的实现类 SubmissionPublisher。本文将先梳理一下接口中具体的处理流程，然后再以几个调用者的例子来帮助大家理解。

JDK9 中的实现

再放上一下上文中的响应式流的交互流程：

订阅者向发布者发送订阅请求。
发布者根据订阅请求生成令牌发送给订阅者。
订阅者根据令牌向发布者发送请求 N 个数据。
发送者根据订阅者的请求数量返回 M(M<=N)个数据
重复 3，4
数据发送完毕后由发布者发送给订阅者结束信号

该流程的角度是以接口调用的交互来说的，而考虑实际的 coding 工作中，我们的调用流程其实为:

创建发布者
创建订阅者
订阅令牌交互
发送信息

接下来我们按照这个流程来梳理一下代码细节。

创建发布者

对于实现响应流的最开始的步骤，便是创建一个发布者。之前提到在 JDK9 中提供了一个发布者的简单实现 SubmissionPublisher。SubmissionPublisher 继承自 Flow.Publisher，他有三种构造函数:

    public SubmissionPublisher() {        this(ASYNC_POOL, Flow.defaultBufferSize(), null);    }        public SubmissionPublisher(Executor executor, int maxBufferCapacity) {        this(executor, maxBufferCapacity, null);    }
    public SubmissionPublisher(Executor executor, int maxBufferCapacity,                               BiConsumer<? super Subscriber<? super T>, ? super Throwable> handler)

复制代码

SubmissionPublisher 将使用Executor作为“线程池”向订阅者发送信息。如果需要需要设置线程池的话可以自己传入，否则的话再无参的构造函数中将默认使用ForkJoinPool类的commonPool()方法获取，即无餐构造方法中的 ASYNC_POOL 静态变量。

SubmissionPublisher 会为每一个订阅者单独的建立一个缓冲空间，其大小由入参maxBufferCapacity决定。默认情况下直接使用Flow.defaultBufferSize()来设置，默认为 256。如果缓冲区满了之后会根据发送信息时候的策略确定是阻塞等待还是抛弃数据。

SubmissionPublisher 会在订阅者发生异常的时候（onNext 处理中），会调用最后一个参数 handler 方法，然后才会取消订阅。默认的时候为 null，也就是不会处理异常。

最简单的创建 SubmissionPublisher 的方法就是直接使用无参构造方法:

SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

复制代码

上文书说到，因为 SubmissionPublisher 实现了 AutoCloseable 接口，所以可以用 try 来进行资源回收可以省略 close()的调用：

try (SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>()){}

复制代码

但是也可以手动的调用 close()方法来显示的关闭发布者，关闭后再发送数据就会抛出异常:

if (complete)    throw new IllegalStateException("Closed");

复制代码

创建订阅者

上文中咱们没有手动创建订阅者，而是直接调用 SubmissionPublisher 中的 consume 方法使用其内部的订阅者来消费消息。在本节可以实现接口Flow.Subscriber<T>创建一个 SimpleSubscriber 类:

public class SimpleSubscriber implements Flow.Subscriber<Integer> {    private Flow.Subscription subscription;    /**     * 订阅者名称     */    private String name;    /**     * 定义最大消费数量     */    private final long maxCount;    /**     * 计数器     */    private long counter;    public SimpleSubscriber(String name, long maxCount) {        this.name = name;        this.maxCount = maxCount <= 0 ? 1 : maxCount;    }    @Override    public void onSubscribe(Flow.Subscription subscription) {        this.subscription = subscription;        System.out.printf("订阅者：%s，最大消费数据： %d。%n", name, maxCount);        // 实际上是等于消费全部数据        subscription.request(maxCount);    }    @Override    public void onNext(Integer item) {        counter++;        System.out.printf("订阅者：%s 接收到数据：%d.%n", name, item);        if (counter >= maxCount) {            System.out.printf("准备取消订阅者： %s。已处理数据个数：%d。%n", name, counter);            // 处理完毕，取消订阅            subscription.cancel();        }    }    @Override    public void onError(Throwable t) {        System.out.printf("订阅者: %s，出现异常： %s。%n", name, t.getMessage());    }    @Override    public void onComplete() {        System.out.printf("订阅者: %s 处理完成。%n", name);    }}

复制代码

SimpleSubscriber 是一个简单订阅者类，其逻辑是根据构造参数可以定义其名称 name 与最大处理数据值 maxCount，最少处理一个数据。

当发布者进行一个订阅的时候会生成一个令牌 Subscription 作为参数调用 onSubscribe 方法。在订阅者需要捕获该令牌作为后续与发布者交互的纽带。一般来说在 onSubscribe 中至少调用一次 request 且参数需要>0，否则发布者将无法向订阅者发送任何信息，这也是为什么 maxCount 需要大于 0。

当发布者开始发送数据后，会异步的调用 onNext 方法并将数据传入。该类中使用了一个计数器对数据数量进行校验，当达到最大值的时候，则会通过令牌(subscription)异步通知发布者订阅结束，然后发送者再异步的调用发订阅者的 onComplete 方法，以处理完成流程。

其中的 onError 和 onComplete 方法只进行打印，这里就不再说了。

以上的这个订阅者可以看作是一个 push 模型的实现，因为当开始订阅时订阅者就约定了需要接受的数量，然后在后续的处理(onNext)中不再请求新数据。

我们可以用以下的代码创建一个名称为 S1，消费 2 个元素的订阅者:

SimpleSubscriber sub1 = new SimpleSubscriber("S1", 2);

复制代码

订阅令牌交互

当我们可以创建了发送者和订阅者之后，我们需要确认一下进行交互的顺序，由于响应流的处理就是对于事件的处理，所以事件的顺序十分重要，具体顺序如下：

我们创建一个发布者publisher一个订阅者subscriber
订阅者subscriber通过调用发布者的subscribe()方法进行信息订阅。如果订阅成功，则发布者将生成一个令牌（Subscription）并作为入参调用订阅者的订阅事件方法onSubscribe()。如果调用异常则会直接调用订阅者的onError错误处理方法，并抛出IllegalStateException异常然后结束订阅。
在onSubscribe()中，订阅者需要通过调用令牌（Subscription）的请求方法request(long)来异步的向发布者请求数据。
当发布者有数据可以发布的时候，则会异步的调用订阅者的onNext()方法，直到所有消息的总数已经满足了订阅者调用request的数据请求上限。所以当订阅者请求订阅的消息数为Long.MAX_VALUE时，实际上是消费所有数据，即 push 模式。如果发布者没有数据要发布了，则可以会调用发布者自己的close()方法并异步的调用所有订阅者的onComplete()方法来通知订阅结束。
发布者可以随时向发布者请求更多的元素请求（一般在 onNext 里），而不用等到之前的处理完毕，一般是与之前的数据数量进行累加。
放发布者遇到异常的时候会调用订阅者的onError()方法。

上面的描述中是只使用的一个订阅者来进行描述的，后面的例子中将说明发布者可以拥有多个订阅者(甚至 0 个订阅者)。

发送信息

当发布者需要推送消息的时候会调用 submit 方法或者 offer 方法，上文中我们提到 submit 实际上是 offer 的一种简单实现，本节咱们自己比较一下。

首先他们的方法签名为:

int offer(T item, long timeout, TimeUnit unit, BiPredicate<Flow.Subscriber<? super T>,? super T> onDrop)int offer(T item, BiPredicate<Flow.Subscriber <? super T>,? super T> onDrop)int submit(T item)

复制代码

而 submit 和 offer 的直接方法为:

    public int submit(T item) {        return doOffer(item, Long.MAX_VALUE, null);    }        public int offer(T item,                     BiPredicate<Subscriber<? super T>, ? super T> onDrop) {    return doOffer(item, 0L, onDrop);

复制代码

可以看到他们的底层调用的都是 doOffer 方法，而doOffer的方法签名为:

    private int doOffer(T item, long nanos,                        BiPredicate<Subscriber<? super T>, ? super T> onDrop)

复制代码

所以我们可以直接看doOffer()方法。doOffer()方法是可选阻塞时长的，而时长根据入参数 nanos 来决定。而onDrop()是一个删除判断器，如果调用BiPredicate的test()方法结果为true则会再次重试（根据令牌中的nextRetry属性与发布器中的retryOffer()方法组合判断，但是具体实现还没梳理明白）；如果结果为flase则直接删除内容。doOffer()返回的结果为正负两种，正数的结果为发送了数据，但是订阅者还未消费的数据(估计值，因为是异步多线程的);如果为负数，则返回的是重拾次数。

所以，根据submit()的参数我们可以发现，submit 会一直阻塞直到数据可以被消费(因为不会阻塞超时，所以不需要传入onDrop()方法)。而我们可以根据需要配置offer()选择器。如果必须要求数据都要被消费的话，那就可以直接选择submit(),如果要设置重试次数的话就可以选择使用offer()

异步调用的例子

下面看一个具体的程序例子，程序将以 3 秒为周期进行数据发布：

public class PeriodicPublisher {
    public static final int WAIT_TIME = 2;    public static final int SLEEP_TIME = 3;
    public static void main(String[] args) {        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();        // 创建4订阅者        SimpleSubscriber subscriber1 = new SimpleSubscriber("S1", 2);        SimpleSubscriber subscriber2 = new SimpleSubscriber("S2", 4);        SimpleSubscriber subscriber3 = new SimpleSubscriber("S3", 6);        SimpleSubscriber subscriber4 = new SimpleSubscriber("S4", 10);        // 前三个订阅者直接进行订阅        publisher.subscribe(subscriber1);        publisher.subscribe(subscriber2);        publisher.subscribe(subscriber3);        // 第四个方法延迟订阅        delaySubscribeWithWaitTime(publisher, subscriber4);        // 开始发送消息        Thread pubThread = publish(publisher, 5);        try {            // 等待处理完成            pubThread.join();        } catch (InterruptedException e) {            e.printStackTrace();        }    }    public static Thread publish(SubmissionPublisher<Integer> publisher, int count) {        Thread t = new Thread(() -> {            IntStream.range(1,count)                    .forEach(item ->{                        publisher.submit(item);                        sleep(item);                    });            publisher.close();        });        t.start();        return t;    }            private static void sleep(Integer item) {        try {            System.out.printf("推送数据：%d。休眠 3 秒。%n", item);            TimeUnit.SECONDS.sleep(SLEEP_TIME);        } catch (InterruptedException e) {            e.printStackTrace();        }    }    private static void delaySubscribeWithWaitTime(SubmissionPublisher<Integer> publisher, Flow.Subscriber<Integer> sub) {        new Thread(() -> {            try {                TimeUnit.SECONDS.sleep(WAIT_TIME);                publisher.subscribe(sub);            } catch (InterruptedException e) {                e.printStackTrace();            }        }).start();    }
}

复制代码

代码后是运行结果如下:

订阅者：S1，最大消费数据： 2。推送数据：1。休眠 3 秒。订阅者：S3，最大消费数据： 6。订阅者：S2，最大消费数据： 4。订阅者：S2 接收到数据：1.订阅者：S3 接收到数据：1.订阅者：S1 接收到数据：1.订阅者：S4，最大消费数据： 10。推送数据：2。休眠 3 秒。订阅者：S2 接收到数据：2.订阅者：S3 接收到数据：2.订阅者：S1 接收到数据：2.订阅者：S4 接收到数据：2.准备取消订阅者： S1。已处理数据个数：2。推送数据：3。休眠 3 秒。订阅者：S4 接收到数据：3.订阅者：S2 接收到数据：3.订阅者：S3 接收到数据：3.推送数据：4。休眠 3 秒。订阅者：S4 接收到数据：4.订阅者：S3 接收到数据：4.订阅者：S2 接收到数据：4.准备取消订阅者： S2。已处理数据个数：4。推送数据：5。休眠 3 秒。订阅者：S3 接收到数据：5.订阅者：S4 接收到数据：5.订阅者: S3 处理完成。订阅者: S4 处理完成。

复制代码

由于是异步执行，所以在“接收数据”部分的顺序可能不同。

我们分析一下程序的执行流程。

创建一个发布者实例
创建四个订阅者实例 S1、S2、S3、S4，可以接收数据的数量分别为:2、4、6、10。
前三个订阅者立即订阅消息。
S4 的订阅者单独创建一个线程等待WAIT_TIME秒(2 秒)之后进行数据的订阅。
新建一个线程来以SLEEP_TIME秒(3 秒)为间隔发布 5 个数据。
将 publish 线程 join()住等待流程结束。

执行的日志满足上述流程而针对一些关键点为:

S4 在发送者推送数据"1"的时候还未订阅，所以 S4 没有接收到数据"1"。
当发送数据"2"的时候 S1 已经接收够了预期数据 2 个，所以取消了订阅。之后只剩下 S2、S3、S4。
当发送数据"4"的时候 S2 已经接收够了预期数据 4 个，所以取消了订阅。之后只剩下 S3、S4。
当发送数据"5"的时候只剩下 S3、S4，当发送完毕后publisher调用close()方法，通知 S3、S4 数据处理完成。

需要注意的是，如果在最后 submit 完毕之后直接 close()然后结束进行的话可能订阅者并不能执行完毕。但是由于在任意一次submit()之后都有一次 3 秒的等待，所以本程序是可以执行完毕的。

最后

本文中的例子是是简单的实现，可以通过调整订阅者中的 request 的参数，与在 onNext 中添加 request 调用来测试背压的效果，还可以将 submit 调整为 offer 并添加 onDrop 方法以观察抛弃信息时的流程。同时本文没有提供 Processor 的例子，各位也可以自行学习。

总结一下流程: 订阅者向发布者进行订阅，然后发布者向订阅者发送令牌。订阅者使用令牌请求消息，发送者根据请求消息的数量推送消息。订阅者可以随时异步追加需要的更多信息。

JDK9 中在Flow接口中实现了 Java API 的 4 个接口，并提供了SubmissionPublisher<T>作为Publisher<T>接口的简单实现。

发布于: 刚刚阅读数: 2

原文链接:【http://xie.infoq.cn/article/61fd85848c194d0e97c2c10ca】。文章转载请联系作者。

蜜糖的代码注释

关注

还未添加个人签名 2020.06.26 加入

欢迎关注同名公众号！

发布

暂无评论

创作场景

JDK9 响应式流使用详解

JDK9 中的实现

创建发布者

创建订阅者

订阅令牌交互

发送信息

异步调用的例子

最后

蜜糖的代码注释

评论