反应式编程

reactive 是一种新的编程思想, 如同名字一样, 反应式编程。而Reactor 是一个工具包，类似于 Spring一样。这点我们可以直接在Spring的官网上可以看到。本篇基于小编自己的学习进行总结。

public interface Publisher<T> {// 绑定一个订阅者public void subscribe(Subscriber<? super T> s);
}

public interface Subscriber<T> {// 确定订阅关系public void onSubscribe(Subscription s);// 处理数据public void onNext(T t);// 错误处理public void onError(Throwable t);// 当事件处理完时触发public void onComplete();
}

public interface Subscription {// 获取指定数量的数据public void request(long n);// 取消订阅关系public void cancel();
}

enum CreateMode {PUSH_ONLY, PUSH_PULL
}

@Test@DisplayName("Flux Pull模式 Integer.MAX_VALUE")public void testFluxPull() {Flux.generate((Consumer<SynchronousSink<Integer>>) sink -> {int k = (int) (Math.random() * 10);sink.next(k);})// 默认获取 request(Integer.MAX_VALUE).subscribe(integer -> System.out.println("Pull:" + integer));}@Test@DisplayName("Flux Pull模式 request调用一次,则调用Sink生产一次")public void testFluxPullTwo() {Flux.generate((Consumer<SynchronousSink<Integer>>) sink -> {int k = (int) (Math.random() * 10);sink.next(k);}).subscribe(new Subscriber<Integer>() {Subscription subscription;private int count;@Overridepublic void onSubscribe(Subscription s) {this.subscription = s;// 订阅时候,生产1条数据this.subscription.request(1);}@Overridepublic void onNext(Integer integer) {count++;System.out.println("处理:" + integer);// 在处理1次，当第二次处理时候,就不拉数据了if (count < 2) {this.subscription.request(1);}}@Overridepublic void onError(Throwable t) {System.out.println("onError");}@Overridepublic void onComplete() {System.out.println("onComplete");}});}

处理:3
处理:1

@Test@DisplayName("Flux Push模式")public void testFluxPush() {Flux.create((Consumer<FluxSink<Integer>>) sink -> {int k = (int) (Math.random() * 10);sink.next(k);}).subscribe(new Subscriber<Integer>() {Subscription subscription;@Overridepublic void onSubscribe(Subscription s) {this.subscription = s;this.subscription.request(1);}@SneakyThrows@Overridepublic void onNext(Integer integer) {System.out.println("处理:" + integer);}@Overridepublic void onError(Throwable t) {System.out.println("处理失败");}@Overridepublic void onComplete() {System.out.println("处理完成");}});}

处理:9

StopWatch '耗时统计': running time = 2070915374 ns
---------------------------------------------
ns         %     Task name
---------------------------------------------
063036666   003%  基于事件驱动的编程思想
2007878708  097%  传统阻塞式的编程思想

@Test@DisplayName("Mono 事件驱动的好处")public void testMono() {StopWatch stopWatch = new StopWatch();stopWatch.start("基于事件驱动的编程思想");Mono<String> userNameMono = getMonoUserName();stopWatch.stop();stopWatch.start("传统阻塞式的编程思想");System.out.println(getUserName());stopWatch.stop();System.out.println(userNameMono.block());System.out.println(stopWatch.prettyPrint());}@SneakyThrowspublic String getUserName() {Thread.sleep(2000L);return "JayChou";}/*** 基于事件驱动的编程思想** @return Mono<String>*/public Mono<String> getMonoUserName() {return Mono.create(monoSink -> {try {Thread.sleep(2000L);} catch (InterruptedException e) {monoSink.error(new RuntimeException(e));return;}monoSink.success("JayChou");});}