js网站特效,给人做网站赚钱吗,南宁高端网站建设,学校网站开发实际意义一、响应式编程概述
1.1 背景知识
为了应对高并发服务器端开发场景#xff0c;在2009 年#xff0c;微软提出了一个更优雅地实现异步编程的方式——Reactive Programming#xff0c;我们称之为响应式编程。随后#xff0c;Netflix 和LightBend 公司提供了RxJava 和Akka S…一、响应式编程概述
1.1 背景知识
为了应对高并发服务器端开发场景在2009 年微软提出了一个更优雅地实现异步编程的方式——Reactive Programming我们称之为响应式编程。随后Netflix 和LightBend 公司提供了RxJava 和Akka Stream 等技术使得Java 平台也有了能够实现响应式编程的框架。 在2017 年9 月28 日Spring 5 正式发布。Spring 5 发布最大的意义在于它将响应式编程技术的普及向前推进了一大步。而同时作为在背后支持Spring 5 响应式编程的框架Spring Reactor也进入了里程碑式的3.1.0 版本。
1.2 什么是响应式编程
响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流而相关的计算模型会自动将变化的值通过数据流进行传播。 响应式编程基于reactorReactor 是一个运行在 Java8 之上的响应式框架的思想当你做一个带有一定延迟的才能够返回的io操作时不会阻塞而是立刻返回一个流并且订阅这个流当这个流上产生了返回数据可以立刻得到通知并调用回调函数处理数据。 电子表格程序就是响应式编程的一个例子。单元格可以包含字面值或类似B1C1的公式而包含公式的单元格的值会依据其他单元格的值的变化而变化。 响应式传播核心特点之一变化传播一个单元格变化之后会像多米诺骨牌一样导致直接和间接引用它的其他单元格均发生相应变化。
1.3 基于 Reactor 实现
Reactor 是一个运行在 Java8 之上满足 Reactice 规范的响应式框架它提供了一组响应式风格的 API。 Reactor 有两个核心类 Flux 和 Mono这两个类都实现 Publisher 接口。 Flux 类似 RxJava 的 Observable它可以触发零到多个事件并根据实际情况结束处理或触发错误。 Mono 最多只触发一个事件所以可以把 Mono 用于在异步任务完成时发出通知。 Flux 和 Mono 都是数据流的发布者使用 Flux 和 Mono 都可以发出三种数据信号元素值错误信号完成信号错误信号和完成信号都代表终止信号终止信号用于告诉订阅者数据流结束了错误信号终止数据流同时把错误信息传递给订阅者。 三种信号的特点 错误信号和完成信号都是终止信号不能共存 如果没有发送任何元素值而是直接发送错误或者完成信号表示是空数据流 如果没有错误信号也没有完成信号表示是无限数据流
Mono 原理图如下
Flux原理图如下
结合上面两个图发现Mono和Flux非常相似。只是Mono只接收一个元素而Flux接收多个元素
二、示例代码
2.1 Mono
package com.reactor.demo;import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Mono;import java.time.Duration;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;Slf4j
public class MonoTest {Testpublic void test1() {//just用法Mono.just(hello world).subscribe(System.out::println);//runnable创建monoMonoVoid sinkMono Mono.fromRunnable(() - System.out.println(runnable));//这句不会输出sinkMono.doOnNext(unused - System.out.println(void success));//这句也不会输出sinkMono.subscribe(o - System.out.println(void result o));//创建一个不包含任何元素只发布结束消息的序列。这里的hello empty是不会输出的。Mono.empty()//输出“empty的入参是null”.doOnSuccess(o - System.out.println(empty的入参是 o))//这句不会输出.subscribe(o - System.out.println(hello empty));//empty里面至少还有一个结束消息而never则是真的啥都没有。never的入参是不会输出 这里的hello never也不会输出Mono.never().doOnSuccess(o - System.out.println(never的入参是 o)).subscribe(o - System.out.println(hello never));}Testpublic void test2() {//传入supplierMono.fromSupplier(() - Hello supplier).subscribe(System.out::println);//传入optionalMono.justOrEmpty(Optional.of(Hello optional)).subscribe(System.out::println);//通过sink来创建一个正常执行的MonoMono.create(sink - sink.success(Hello sink)).subscribe(System.out::println);//通过sink来创建一个抛出异常的MonoMono.create(sink - sink.error(new RuntimeException(sink error))).subscribe(System.out::println);//defer的入参实际上是一个Mono工厂Mono.defer(() - Mono.just(hello defer)).subscribe(System.out::println);}Testpublic void test3() {//callable有返回值Mono.fromCallable(() - callable).subscribe(System.out::println);//runnable无返回值MonoVoid mono Mono.fromRunnable(() - System.out.println(run));//下面的hello runnable是不会输出的。因为subscribe一个MonoVoid,不会产生任何结果mono.subscribe(o - System.out.println(hello runnable));}Testpublic void test4() {//延迟3秒输出Mono.delay(Duration.ofSeconds(3)).doOnNext(new ConsumerLong() {Overridepublic void accept(Long aLong) {System.out.println(aLong);}}).block();}Testpublic void test5() {//直接输出了异常Mono.error(new RuntimeException(这是一个异常)).subscribe(new ConsumerObject() {Overridepublic void accept(Object o) {System.out.println(error: o);}});Mono.defer(() - {return Mono.error(new RuntimeException(这是第二个异常));}).subscribe(new ConsumerObject() {Overridepublic void accept(Object o) {System.out.println(defer error: o);}});}Testpublic void test6() {//通过map可以对元素进行转换Mono.just(just one).map(new FunctionString, Integer() {Overridepublic Integer apply(String s) {return 1;}}).doOnNext(new ConsumerInteger() {Overridepublic void accept(Integer integer) {System.out.println(转换后的结果 integer);}}).subscribe();}
}2.1 Flux
package com.reactor.demo;import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import java.time.Duration;
import java.util.Arrays;
import java.util.function.Consumer;public class FluxTest {/*** 基本用法*/Testpublic void test1() {//通过just传入可变的参数依次输出Flux.just(hello, world, just).doOnNext(System.out::println).doOnComplete(() - System.out.println(just over)).subscribe();//传入一个范围Flux.range(100, 10).doOnNext(System.out::println).doOnComplete(() - System.out.println(OK)).subscribe();//传入listFlux.fromIterable(Arrays.asList(01, 02, 03)).doOnNext(System.out::println).subscribe();//传入一个数组Flux.fromArray(new Object[]{obj1, obj2}).doOnNext(System.out::println).subscribe();}/*** 处理空值*/Testpublic void testEmpty() {//如果序列是个空的就给个默认值Flux.empty().defaultIfEmpty(1).doOnNext(System.out::println).subscribe();//如果序列是空的就用新序列代替Flux.empty().switchIfEmpty(Mono.just(100)).doOnNext(System.out::println).subscribe();}/*** 序列在执行时的一些监听方法doOnXXXX*/Testpublic void testDoOn() {System.out.println(----------);Flux.range(100, 10).doOnNext(System.out::println).doOnComplete(() - System.out.println(OK));System.out.println(----------);Flux.range(100, 10).doFirst(() - System.out.println(第一个执行开始)).subscribe();System.out.println(----------);Flux.range(100, 10).doFinally(it - System.out.println(终止信号的类型为 it.name())).subscribe();System.out.println(----------);Flux.range(100, 10).doOnSubscribe(it - System.out.println(该序列已被订阅)).subscribe();System.out.println(----------);Flux.range(100, 10).doOnRequest(value - System.out.println(doOnRequest: value)).subscribe();//在完成或者error时也就是序列终止时执行runnableSystem.out.println(----------);Flux.range(100, 10).doOnTerminate(() - System.out.println(doOnTerminate)).subscribe();//doOnEach每次向下游传播都会得到一个信号类型可以根据该信号类型执行一些操作System.out.println(----------);Flux.range(100, 10).doOnEach(it - System.out.println(doOnEach: it)).subscribe();}/*** filter用法*/Testpublic void testFilter() {System.out.println(----------);//将上游的数据进行类型判断符合该类型的数据将流向下游Flux.just(new Object(), Hello, 1).ofType(String.class).doOnNext(System.out::println).doOnComplete(() - System.out.println(过滤String示例)).subscribe();System.out.println(----------);//过滤数据Flux.range(100, 10).filter(it - it 105).doOnComplete(() - System.out.println(取出大于105示例)).subscribe();System.out.println(----------);//将重复数据过滤,重复数据在整个序列中只保留一个Flux.range(100, 10).concatWith(Flux.just(100, 100, 100)).distinct().doOnNext(System.out::println).doOnComplete(() - System.out.println(去除重复数字示例)).subscribe();System.out.println(----------);//将后来的重复数据过滤如下第二个flux拼接到第一个序列时只会把第二个元素本身的重复元素过滤Flux.range(100, 10).concatWith(Flux.just(100, 100, 100)).distinctUntilChanged().doOnNext(System.out::println).doOnComplete(() - System.out.println(将后来的重复数据过滤)).subscribe();System.out.println(----------);//在序列的开始获取5个元素// limitRequest为true时则不管该序列会发射多少元素该参数会向上传递背压则上游序列只会发出设定的5个元素//为false时则不控制上有元素可以发出N个元素Flux.range(100, 10).take(5, false).doOnComplete(() - System.out.println(在序列的开始获取5个元素)).subscribe();System.out.println(----------);//参数为时间单位意味着take获取元素只会在该时间限制内获取。Flux.range(100, 10).take(Duration.ofSeconds(10)).doOnNext(new ConsumerInteger() {Overridepublic void accept(Integer integer) {System.out.println(当前时间戳为 System.currentTimeMillis() 数字为 integer);}}).doOnComplete(() - System.out.println(在指定时间内获取元素)).subscribe(System.out::println);System.out.println(----------);//获取最后的N位元素Flux.range(100, 10).takeLast(2).doOnComplete(() - System.out.println(获取最后的2位元素)).subscribe(System.out::println);System.out.println(----------);//获取元素知道符合条件后停止向下游发送数据,包括条件本身也就是当it105的元素也会被发布至下游Flux.range(100, 10).takeUntil(it - it 105).doOnComplete(() - System.out.println(一直取数直到大于105结束)).subscribe(System.out::println);System.out.println(----------);//获取元素当元素符合该断言时如果不符合直接终止,不包含条件本身Flux.range(100, 10).takeWhile(it - it 105).doOnComplete(() - System.out.println(取出小于105示例)).subscribe(System.out::println);System.out.println(----------);//获取指定某个位置的一个元素Flux.range(100, 10).elementAt(0).doOnSuccess(new ConsumerInteger() {Overridepublic void accept(Integer i) {System.out.println(获取指定某个位置的一个元素: i);}}).subscribe();System.out.println(----------);//获取最后一个元素last()如果为空则抛出异常last(1)如果为空则发出默认值Flux.range(100, 10).takeWhile(it - it 105).last(1).subscribe(System.out::println);System.out.println(----------);//跳至第几秒开始执行Flux.range(100, 10).skip(Duration.ofSeconds(5)).subscribe(System.out::println);System.out.println(----------);//跳至第几个元素开始执行Flux.range(100, 10).skip(5).subscribe(System.out::println);System.out.println(----------);//从开始跳到最后第N个元素结束Flux.range(100, 10).skipLast(5).subscribe(System.out::println);System.out.println(----------);//跳至满足条件的地方开始执行,从第一个元素开始知道满足条件开始发送至下游Flux.range(100, 10).skipUntil(it - it 105).subscribe(System.out::println);System.out.println(----------);//每隔一段时间抽取样本数(取在这个时间的最后一个元素)如果相隔实现大于序列的执行时间则去最后一元素Flux.range(100, 100000000).sample(Duration.ofMillis(100)).subscribe(System.out::println);System.out.println(----------);//每隔一段时间抽取样本数(取在这个时间的第一个元素)如果相隔实现大于序列的执行时间则取第一个元素Flux.range(100, 10).sampleFirst(Duration.ofMillis(100)).subscribe(System.out::println);System.out.println(----------);//只获取一个元素single()如果为空或者超多一个抛出异常single(1)如果为空返回默认值如果多个抛出异常,singleOrEmpty()可以允许为空Flux.range(100, 10).single(1).subscribe(System.out::println);}/*** 当被订阅后如果发生异常则stream会停止运行* 此时可以通过处理error来决定如何处理异常* 可以将异常跳过、将异常替换等*/Testpublic void testErrorHandle() {System.out.println(----------);Flux.just(1, 2, 3, 0, 5, 4).map(it - {it 100 / it;return it;})//报错后返回并停止运行.onErrorResume(e - {return Mono.just(10000);}).doFinally(type - {System.out.println(type);}).subscribe(System.out::println);System.out.println(----------);Flux.just(1, 2, 3).doOnNext(new ConsumerInteger() {Overridepublic void accept(Integer integer) {System.out.println(integer);if (integer 2) {throw new RuntimeException(触发异常);}}}).doOnError(new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) {System.out.println(doOnError throwable.getMessage());}}).subscribe();System.out.println(----------);Flux.just(1, 2, 3, 0, 5, 4).map(it - {it 100 / it;return it;})//报错后继续运行并执行相关操作.onErrorContinue((e, it) - {System.out.println(e.getMessage());}).doFinally(type - {System.out.println(type);}).subscribe(System.out::println);}Testpublic void flatMapTest() {//输出50100Flux.just(5, 10).flatMap(x - Flux.just(x * 10)).toStream().forEach(System.out::println);}
}Mono源码分析
以下面的代码为例来分析一下Mono源码
Test
public void test0() {//just用法Mono.just(hello world).subscribe(new ConsumerString() {Overridepublic void accept(String s) {System.out.println(accept: s);}});
}Mono.just返回了个啥
原来是静态方法返回了个MonoJust对象入参作为构造参数传入 public static T MonoT just(T data) {return onAssembly(new MonoJust(data));}MonoJust原来是继承Mono的 有两个要点 1、有一个value字段来保存入参 2、一个subscribe方法执行了什么动作后面再分析
final class MonoJustT
extends MonoTimplements Fuseable.ScalarCallableT, Fuseable, SourceProducerT {final T value;MonoJust(T value) {this.value Objects.requireNonNull(value, value);}Overridepublic T call() throws Exception {return value;}Overridepublic T block(Duration m) {return value;}Overridepublic T block() {return value;}Overridepublic void subscribe(CoreSubscriber? super T actual) {actual.onSubscribe(Operators.scalarSubscription(actual, value));}Overridepublic Object scanUnsafe(Attr key) {if (key Attr.BUFFERED) return 1;if (key Attr.RUN_STYLE) return Attr.RunStyle.SYNC;return null;}
}Mono.subscribe到底做了什么
subscribe方法入参是一个Consumer对象这里可以理解为一个回调方法 public final Disposable subscribe(Consumer? super T consumer) {Objects.requireNonNull(consumer, consumer);return subscribe(consumer, null, null);}继续往下调用了三个参数的subscribe方法 public final Disposable subscribe(Nullable Consumer? super T consumer,Nullable Consumer? super Throwable errorConsumer,Nullable Runnable completeConsumer) {return subscribe(consumer, errorConsumer, completeConsumer, (Context) null);}这里除了我们入参的第一个consumer还有另外两个consumer。还有一个context即上下文这里也是空的。 errorConsumer在出错的时候的回调方法 completeConsumer完成时的回调方法这里是一个Runnable public final Disposable subscribe(Nullable Consumer? super T consumer,Nullable Consumer? super Throwable errorConsumer,Nullable Runnable completeConsumer,Nullable Context initialContext) {return subscribeWith(new LambdaMonoSubscriber(consumer, errorConsumer,completeConsumer, null, initialContext));}这里将几个consumer封装成了一个LambdaMonoSubscriber。 接着往下看subscribeWith方法 public final E extends Subscriber? super T E subscribeWith(E subscriber) {subscribe(subscriber);return subscriber;}继续往下分析
OverrideSuppressWarnings(unchecked)public final void subscribe(Subscriber? super T actual) {CorePublisher publisher Operators.onLastAssembly(this);CoreSubscriber subscriber Operators.toCoreSubscriber(actual);//省略部分代码publisher.subscribe(subscriber);}catch (Throwable e) {Operators.reportThrowInSubscribe(subscriber, e);return;}}省略了部分代码publisher.subscribe(subscriber) publisher 即当前mono对象MonoJust实现了这个方法 这里又回到MonoJust里了
Override
public void subscribe(CoreSubscriber? super T actual) {actual.onSubscribe(Operators.scalarSubscription(actual, value));
}这里的actual是什么是上面的LambdaMonoSubscriber 这里又把actual和value封装成Operators.scalarSubscription 看一下LambdaMonoSubscriber定义 LambdaMonoSubscriber主要就是定义了一系列consumer即回调勾子
final class LambdaMonoSubscriberT implements InnerConsumerT, Disposable {final Consumer? super T consumer;final Consumer? super Throwable errorConsumer;final Runnable completeConsumer;final Consumer? super Subscription subscriptionConsumer;final Context initialContext;volatile Subscription subscription;
}看一下LambdaMonoSubscriber.onSubscribe方法 Overridepublic final void onSubscribe(Subscription s) {if (Operators.validate(subscription, s)) {this.subscription s;if (subscriptionConsumer ! null) {try {subscriptionConsumer.accept(s);}catch (Throwable t) {Exceptions.throwIfFatal(t);s.cancel();onError(t);}}else {s.request(Long.MAX_VALUE);}}}我们的示例会进入s.request(Long.MAX_VALUE);这个逻辑 这个s是什么Operators.scalarSubscription 再来看看request方法 Overridepublic void request(long n) {if (validate(n)) {if (ONCE.compareAndSet(this, 0, 1)) {Subscriber? super T a actual;a.onNext(value);if(once ! 2) {a.onComplete();}}}}即调用了LambdaMonoSubscriber的onNext和complete方法 LambdaMonoSubscriber.next
Overridepublic final void onNext(T x) {Subscription s S.getAndSet(this, Operators.cancelledSubscription());if (s Operators.cancelledSubscription()) {Operators.onNextDropped(x, this.initialContext);return;}if (consumer ! null) {try {consumer.accept(x);}catch (Throwable t) {Exceptions.throwIfFatal(t);s.cancel();doError(t);}}if (completeConsumer ! null) {try {completeConsumer.run();}catch (Throwable t) {Operators.onErrorDropped(t, this.initialContext);}}}这里主要是调用了consumer.accept(x);。这个consumer即我们最开始入参的那个回调方法 onComplete同理即运行completeConsumer这个runnable Overridepublic final void onComplete() {Subscription s S.getAndSet(this, Operators.cancelledSubscription());if (s Operators.cancelledSubscription()) {return;}if (completeConsumer ! null) {try {completeConsumer.run();}catch (Throwable t) {Operators.onErrorDropped(t, this.initialContext);}}}如果出错的情况下会执行 Overridepublic final void onError(Throwable t) {Subscription s S.getAndSet(this, Operators.cancelledSubscription());if (s Operators.cancelledSubscription()) {Operators.onErrorDropped(t, this.initialContext);return;}doError(t);}void doError(Throwable t) {if (errorConsumer ! null) {errorConsumer.accept(t);}else {Operators.onErrorDropped(Exceptions.errorCallbackNotImplemented(t), this.initialContext);}}到这里就分析完了
总结一下
Mono.Just构造了一个MonoJust对象用于接收入参value。Mono.subscribe方法会接收一个Consumer可以理解为回调方法Mono.subscribe方法最终会调用LambdaMonoSubscriber里面的onNext方法。onNext执行的即我们传入的回调consumer在正常执行完时会执行LambdaMonoSubscriber中的onComplete。这个runnable也是通过入参传过来的。在执行异常时会执行LambdaMonoSubscriber中的onError。这个回调方法也是通过入参传过来的。
体会一下下面的过程最好调试一下 Testpublic void test0() {//just用法Mono.just(hello world).doOnNext(new ConsumerString() {Overridepublic void accept(String s) {System.out.println(doOnNext1:s);}}).doOnNext(new ConsumerString() {Overridepublic void accept(String s) {System.out.println(doOnNext2:s);}}).subscribe(new ConsumerString() {Overridepublic void accept(String s) {System.out.println(subscribe: s);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) {System.out.println(subscribe exception: throwable.getMessage());}}, new Runnable() {Overridepublic void run() {System.out.println(subscribe complete);}});}doOnNext1:hello world
doOnNext2:hello world
subscribe:hello world
subscribe complete执行流程如下
参考文章
