189 8069 5689
这篇文章主要介绍“响应式编程简介之如何掌握Reactor”，在日常操作中，相信很多人在响应式编程简介之如何掌握Reactor问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”响应式编程简介之如何掌握Reactor”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
成都创新互联专注为客户提供全方位的互联网综合服务，包含不限于成都网站制作、成都网站设计、新平网络推广、微信小程序开发、新平网络营销、新平企业策划、新平品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等，从售前售中售后，我们都将竭诚为您服务，您的肯定，是我们最大的嘉奖；成都创新互联为所有大学生创业者提供新平建站搭建服务，24小时服务热线：13518219792，官方网址：www.cdcxhl.com
Reactor是基于JVM的非阻塞API，他直接跟JDK8中的API相结合，比如：CompletableFuture，Stream和Duration等。
它提供了两个非常有用的异步序列API：Flux和Mono，并且实现了Reactive Streams的标准。
并且还可以和reactor-netty相结合，作为一些异步框架的底层服务，比如我们非常熟悉的Spring MVC 5中引入的WebFlux。
我们知道WebFlux的底层使用的是reactor-netty，而reactor-netty又引用了Reactor。所以，如果你在POM中引入了webFlux依赖：
org.springframework.boot spring-boot-starter-webflux
那么项目将会自动引入Reactor。
如果你用的不是Spring webflux，没关系，你可以直接添加下面的依赖来使用Reactor：
io.projectreactor reactor-core
最最开始的时候微软为.NET平台创建了Reactive Extensions (Rx) library。接着RxJava实现了JVM平台的Reactive。
然后Reactive Streams标准出现了，它定义了Java平台必须满足的的一些规范。并且已经集成到JDK9中的java.util.concurrent类中。
在Flow中定义了实现Reactive Streams的四个非常重要的组件，分别是Publisher，Subscriber，Subscription和Processor。
一般来说reactive在面向对象的编程语言中是以观察者模式的扩展来使用的。
我们来具体看一下这个观察者模式的实现，以Publisher和Subscriber为例：
public static interface Publisher{ public void subscribe(Subscriber subscriber); }
public static interface Subscriber{ public void onSubscribe(Subscription subscription); public void onNext(T item); public void onError(Throwable throwable); public void onComplete(); }
上面定义了两个接口，Publisher和Subscriber，Publisher的作用就是subscribe到subscriber。
而subscriber定义了4个on方法，用来触发特定的事件。
那么Publisher中的subscribe是怎么触发Subscriber的onSubscribe事件呢？
很简单，我们看一个具体的实现：
public void subscribe(Flow.Subscriber subscriber) {
Subscription sub;
if (throwable != null) {
assert iterable == null : "non-null iterable: " + iterable;
sub = new Subscription(subscriber, null, throwable);
} else {
assert throwable == null : "non-null exception: " + throwable;
sub = new Subscription(subscriber, iterable.iterator(), null);
}
subscriber.onSubscribe(sub);
if (throwable != null) {
sub.pullScheduler.runOrSchedule();
}
}上面的例子是PullPublisher的subscribe实现。我们可以看到，在这个subscribe中触发了subscriber.onSubscribe方法。而这就是观察者模式的秘密。
或者说，当Publisher调用subscribe的时候，是主动push subscriber的onSubscribe方法。
熟悉Iterable-Iterator模式的朋友应该都知道，Iterator模式，其实是一个主动的pull模式，因为需要不断的去调用next()方法。所以它的控制权是在调用方。
在现代应用程序中，随着用户量的增多，程序员需要考虑怎么才能提升系统的处理能力。
传统的block IO的方式，因为需要占用大量的资源，所以是不适合这样的场景的。我们需要的是NO-block IO。
JDK中提供了两种异步编程的模型：
第一种是Callbacks，异步方法可以通过传入一个Callback参数的形式来在Callback中执行异步任务。比较典型的像是java Swing中的EventListener。
第二中就是使用Future了。我们使用Callable来提交一个任务，然后通过Future来拿到它的运行结果。
这两种异步编程会有什么问题呢？
callback的问题就在于回调地狱。熟悉JS的朋友应该很理解这个回调地狱的概念。
简单点讲，回调地狱就是在callback中又使用了callback，从而造成了这种callback的层级调用关系。
而Future主要是对一个异步执行的结果进行获取，它的 get()实际上是一个block操作。并且不支持异常处理，也不支持延迟计算。
当有多个Future的组合应该怎么处理呢？JDK8 实际上引入了一个CompletableFuture类，这个类是Future也是一个CompletionStage，CompletableFuture支持then的级联操作。不过CompletableFuture提供的方法不是那么的丰富，可能满足不了我的需求。
于是我们的Reactor来了。
Reactor提供了两个非常有用的操作，他们是 Flux 和 Mono。 其中Flux 代表的是 0 to N 个响应式序列，而Mono代表的是0或者1个响应式序列。
我们看一个Flux是怎么transfer items的：
先看下Flux的定义：
public abstract class Fluximplements Publisher
可以看到Flux其实就是一个Publisher，用来产生异步序列。
Flux提供了非常多的有用的方法，来处理这些序列，并且提供了completion和error的信号通知。
相应的会去调用Subscriber的onNext, onComplete, 和 onError 方法。
我们看下Mono是怎么transfer items的：
看下Mono的定义：
public abstract class Monoimplements Publisher
Mono和Flux一样，也是一个Publisher，用来产生异步序列。
Mono因为只有0或者1个序列，所以只会触发Subscriber的onComplete和onError方法，没有onNext。
另一方面，Mono其实可以看做Flux的子集，只包含Flux的部分功能。
Mono和Flux是可以互相转换的，比如Mono#concatWith(Publisher)返回一个Flux，而 Mono#then(Mono)返回一个Mono.
我们看下Flux创建的例子：
Fluxseq1 = Flux.just("foo", "bar", "foobar"); List iterable = Arrays.asList("foo", "bar", "foobar"); Flux seq2 = Flux.fromIterable(iterable); Flux numbersFromFiveToSeven = Flux.range(5, 3);
可以看到Flux提供了很多种创建的方式，我们可以自由选择。
再看看Flux的subscribe方法：
Disposable subscribe(); Disposable subscribe(Consumer consumer); Disposable subscribe(Consumer consumer, Consumer errorConsumer); Disposable subscribe(Consumer consumer, Consumer errorConsumer, Runnable completeConsumer); Disposable subscribe(Consumer consumer, Consumer errorConsumer, Runnable completeConsumer, Consumer subscriptionConsumer);
subscribe可以一个参数都没有，也可以多达4个参数。
看下没有参数的情况:
FluxnumbersFromFiveToSeven = Flux.range(5, 3); numbersFromFiveToSeven.subscribe();
注意，没有参数并不表示Flux的对象不被消费，只是不可见而已。
看下带参数的情况：consumer用来处理on each事件，errorConsumer用来处理on error事件，completeConsumer用来处理on complete事件，subscriptionConsumer用来处理on subscribe事件。
前面的3个参数很好理解，我们来举个例子：
Fluxints3 = Flux.range(1, 4); ints3.subscribe(System.out::println, error -> System.err.println("Error " + error), () -> System.out.println("Done"), sub -> sub.request(2));
我们构建了从1到4的四个整数的Flux，on each就是打印出来，如果中间有错误的话，就输出Error，全部完成就输出Done。
那么最后一个subscriptionConsumer是做什么用的呢？
subscriptionConsumer accept的是一个Subscription对象，我们看下Subscription的定义：
public interface Subscription {
public void request(long n);
public void cancel();
}Subscription 定义了两个方法，用来做初始化用的，我们可以调用request(n)来决定这次subscribe获取元素的最大数目。
比如上面我们的例子中，虽然构建了4个整数，但是最终输出的只有2个。
上面所有的subscribe方法，都会返回一个Disposable对象，我们可以通过Disposable对象的dispose()方法，来取消这个subscribe。
Disposable只定义了两个方法：
public interface Disposable {
void dispose();
default boolean isDisposed() {
return false;
}dispose的原理是向Flux 或者 Mono发出一个停止产生新对象的信号，但是并不能保证对象产生马上停止。
有了Disposable，当然要介绍它的工具类Disposables。
Disposables.swap() 可以创建一个Disposable，用来替换或者取消一个现有的Disposable。
Disposables.composite(…)可以将多个Disposable合并起来，在后面统一做处理。
到此，关于“响应式编程简介之如何掌握Reactor”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！