[译]使用JDK 9 Flow API进行响应式编程
什么是响应式编程？
响应式编程是关于处理数据项的异步流，也就是应用程序在数据项发生时对其进行响应。 数据流实质上是指随时间发生的数据项序列。与迭代内存数据相比， 这个模型的内存效率更高，因为数据是以流的形式处理的。
在响应式编程模型中，有一个Publisher和一个Subscriber。 Publisher发布一个数据流，Subscriber异步订阅。
该模型还提供了一种机制来引入更高阶的函数，这样就可以使用Processor在流上操作。 Processor转换数据流不需要更改Publisher或Subscriber。 Processor（或一连串的Processor）位于Publisher和Subcriber之间来将一个数据流转换为另一个。 Publisher和Subcriber独立于发生在数据流中的转换。
为什么要响应式编程？
- 更简洁的代码，使其更具可读性。
- 从模板代码中抽离出来，专注于业务逻辑。
- 从底层的线程，同步和并发问题中抽离出来。
- 流处理意味着高效的内存
- 该模型几乎可以应用于任何类型的问题。
JDK 9 Flow API
JDK 9中的Flow API与Reactive Streams Specification相对应，这是一个事实上的标准。 Reactive Streams Specification是标准化响应式编程的举措之一。 已经有几个实现支持Reactive Streams Specification。
Flow API（和Reactive Streams API）在某些方面是迭代器模式和观察者模式的组合。 迭代器模式是一个拉式（pull）的模型，应用程序从数据源拉取项目。 观察者模式是一个推送（push）模型，数据源的项目被推送到应用程序。 使用Flow API，应用程序最初请求（拉）N个项目，然后发布者将最多N个项目推送给订阅者。 所以说它是拉和推编程模型的混合。
过一遍Flow API 接口
@FunctionalInterface
public static interface Flow.Publisher<T> {
public void subscribe(Flow.Subscriber<? super T> subscriber);
}
public static interface Flow.Subscriber<T> {
public void onSubscribe(Flow.Subscription subscription);
public void onNext(T item);
public void onError(Throwable throwable);
public void onComplete();
}
public static interface Flow.Subscription {
public void request(long n);
public void cancel();
}
public static interface Flow.Processor<T,R> extends Flow.Subscriber<T>, Flow.Publisher<R> {
}
订阅者Subscriber
Subscriber订阅Publisher的回调。 除非有请求，数据项目是不会被推送到订阅者，但可能会请求多个项目。 对于给定订阅（Subscription），调用Subscriber的方法是严格按顺序的。 应用程序可以响应订阅者上的以下回调。
onSubscribe
对于给定的订阅，在调用任何其他Subcriber方法之前调用此方法。
onNext
订阅下一个项目调用此方法
onError
在Publisher或Subcriber遇到不可恢复的错误时调用此方法，之后Subscription不会再调用Subscriber其他的方法。
如果Publisher遇到不允许将项目发送给Subscriber的错误，则Subscriber会收到onError消息，然后不会再收到其他消息。
onComplete
当已知不会再额外调用Subscriber的方法，且没有发生有错误而导致终止订阅，调用此方法。之后Subscription不会调用其它Subscriber的方法。
当知道没有更多的消息发送给它时，订阅者收到onComplete。
Subscriber示例
import java.util.concurrent.Flow.*;
...
public class MySubscriber<T> implements Subscriber<T> {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(1); //a value of Long.MAX_VALUE may be considered as effectively unbounded
}
@Override
public void onNext(T item) {
System.out.println("Got : " + item);
subscription.request(1); //a value of Long.MAX_VALUE may be considered as effectively unbounded
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("Done");
}
}
发布者Publisher
发布者将数据流发布给注册的订阅者。 它通常使用Excutor异步发布项目给订阅者。 Publisher确保每个订阅的Subcriber方法严格按顺序调用。
使用JDK的SubmissionPublisher将数据流发布给订阅者的示例
import java.util.concurrent.SubmissionPublisher;
...
//Create Publisher
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//Register Subscriber
MySubscriber<String> subscriber = new MySubscriber<>();
publisher.subscribe(subscriber);
//Publish items
System.out.println("Publishing Items...");
String[] items = {"1", "x", "2", "x", "3", "x"};
Arrays.asList(items).stream().forEach(i -> publisher.submit(i));
publisher.close();
订阅Subscription
连接Flow.Publisher和Flow.Subscriber。 Subscriber只有在请求时才会收到项目，并可能随时通过Subscription取消订阅。
方法
- request：将给定数量的n个项目添加到当前未完成的此订阅需求中。
- cancel：导致Subscriber（最终）停止接收消息。
处理器Processor
充当Subscriber和Publisher的组件。 处理器位于Publisher和Subscriber之间，它把一个流转换为另一个。 可能有一个或多个链接在一起的处理器，链中最后处理器的结果由Subscriber处理。 JDK没有提供任何具体的处理器，因此需要单独编写任何需要的处理器。
将字符串转换为整数的处理器示例
import java.util.concurrent.Flow.*;
import java.util.concurrent.SubmissionPublisher;
...
public class MyTransformProcessor<T,R> extends SubmissionPublisher<R> implements Processor<T, R> {
private Function function;
private Subscription subscription;
public MyTransformProcessor(Function<? super T, ? extends R> function) {
super();
this.function = function;
}
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(1);
}
@Override
public void onNext(T item) {
submit((R) function.apply(item));
subscription.request(1);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
close();
}
}
使用处理器转换数据流的示例代码
import java.util.concurrent.SubmissionPublisher;
...
//Create Publisher
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
//Create Processor and Subscriber
MyFilterProcessor<String, String> filterProcessor = new MyFilterProcessor<>(s -> s.equals("x"));
MyTransformProcessor<String, Integer> transformProcessor = new MyTransformProcessor<>(s -> Integer.parseInt(s));
MySubscriber<Integer> subscriber = new MySubscriber<>();
//Chain Processor and Subscriber
publisher.subscribe(filterProcessor);
filterProcessor.subscribe(transformProcessor);
transformProcessor.subscribe(subscriber);
System.out.println("Publishing Items...");
String[] items = {"1", "x", "2", "x", "3", "x"};
Arrays.asList(items).stream().forEach(i -> publisher.submit(i));
publisher.close();
背压
当Publisher以比Subscriber所消费数据快得多的速率产生背压。 未处理项目的缓冲区的大小可能受到限制。 Flow API没有提供任何API来发信号或处理背压，但是可以有多种策略由自己来处理背压。 看看RxJava如何处理背压。
总结
将响应式编程的API添加到JDK 9是一个好的开始。 许多其他产品也开始提供响应式编程的API来访问其功能。 尽管Flow API允许程序员开始编写响应式程序，但是生态系统仍然需要发展。
例如，响应式程序可能仍然使用传统的API来访问数据库，也许是因为并不是所有的数据库都支持用于响应式编程的API。 即响应式程序依赖的API可能还不支持响应式编程模型。
