共计 7340 个字符，预计需要花费 19 分钟才能阅读完成。
JDK9 响应式流应用详解
上文中咱们简略提到了 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> 接口的简略实现。
关键词：java 培训