免费网站推广服务,做网站推广话术,网站名称管理,百度实名认证1. Reactor是什么
Reactor 是一个基于Reactive Streams规范的响应式编程框架。它提供了一组用于构建异步、事件驱动、响应式应用程序的工具和库。Reactor 的核心是 Flux#xff08;表示一个包含零到多个元素的异步序列#xff09;和 Mono表示一个包含零或一个元素的异步序列…1. Reactor是什么
Reactor 是一个基于Reactive Streams规范的响应式编程框架。它提供了一组用于构建异步、事件驱动、响应式应用程序的工具和库。Reactor 的核心是 Flux表示一个包含零到多个元素的异步序列和 Mono表示一个包含零或一个元素的异步序列。Reactor 通过提供响应式的操作符如map、filter、flatMap等使得开发者能够方便地进行数据流的转换和处理。
2. Reactor、Callback、CompletableFuture三种形式异步编码对比
编码简洁程度Reactor最优Reactor线程利用率最高因实现了Reactive Streams规范拥有背压事件驱动特性此处暂不展开)
代码如下 pom依赖 dependencyManagementdependenciesdependencygroupIdio.projectreactor/groupIdartifactIdreactor-bom/artifactIdversion2023.0.0/versiontypepom/typescopeimport/scope/dependency/dependencies
/dependencyManagementdependenciesdependencygroupIdio.projectreactor/groupIdartifactIdreactor-core/artifactId/dependency
/dependenciesCallback回调地狱
interface FirstCallback {void onCompleteFirst(String result);void onErrorFirst(Exception e);
}interface SecondCallback {void onCompleteSecond(String result);void onErrorSecond(Exception e);
}interface ThirdCallback {void onCompleteThird(String result);void onErrorThird(Exception e);
}class AsyncOperations {static void firstOperation(FirstCallback firstCallback) {new Thread(() - {try {// 模拟异步操作Thread.sleep(2000);// 操作完成后调用回调函数firstCallback.onCompleteFirst(First operation completed);} catch (Exception e) {// 发生异常时调用错误回调firstCallback.onErrorFirst(e);}}).start();}static void secondOperation(String input, SecondCallback secondCallback) {new Thread(() - {try {// 模拟异步操作Thread.sleep(2000);// 操作完成后调用回调函数secondCallback.onCompleteSecond(Second operation completed with input: input);} catch (Exception e) {// 发生异常时调用错误回调secondCallback.onErrorSecond(e);}}).start();}static void thirdOperation(String input, ThirdCallback thirdCallback) {new Thread(() - {try {// 模拟异步操作Thread.sleep(2000);// 操作完成后调用回调函数thirdCallback.onCompleteThird(Third operation completed with input: input);} catch (Exception e) {// 发生异常时调用错误回调thirdCallback.onErrorThird(e);}}).start();}
}public class CallbackHellExample {public static void main(String[] args) {AsyncOperations.firstOperation(new FirstCallback() {Overridepublic void onCompleteFirst(String result) {System.out.println(First Callback: result);// 第一次操作完成后调用第二次操作AsyncOperations.secondOperation(result, new SecondCallback() {Overridepublic void onCompleteSecond(String result) {System.out.println(Second Callback: result);// 第二次操作完成后调用第三次操作AsyncOperations.thirdOperation(result, new ThirdCallback() {Overridepublic void onCompleteThird(String result) {System.out.println(Third Callback: result);}Overridepublic void onErrorThird(Exception e) {System.out.println(Error in Third Callback: e.getMessage());}});}Overridepublic void onErrorSecond(Exception e) {System.out.println(Error in Second Callback: e.getMessage());}});}Overridepublic void onErrorFirst(Exception e) {System.out.println(Error in First Callback: e.getMessage());}});// 主线程继续执行其他操作System.out.println(Main thread continues...);}
}CompletableFuture优化Callback回调地狱
public class CompletableFutureExample {public static void main(String[] args) {CompletableFutureString firstOperation CompletableFuture.supplyAsync(() - {try {// 模拟异步操作Thread.sleep(2000);return First operation completed;} catch (InterruptedException e) {throw new RuntimeException(e);}});CompletableFutureString secondOperation firstOperation.thenApplyAsync(result - {System.out.println(First CompletableFuture: result);try {// 模拟异步操作Thread.sleep(2000);return Second operation completed with input: result;} catch (InterruptedException e) {throw new RuntimeException(e);}});CompletableFutureString thirdOperation secondOperation.thenApplyAsync(result - {System.out.println(Second CompletableFuture: result);try {// 模拟异步操作Thread.sleep(2000);return Third operation completed with input: result;} catch (InterruptedException e) {throw new RuntimeException(e);}});thirdOperation.whenComplete((result, throwable) - {if (throwable null) {System.out.println(Third CompletableFuture: result);} else {System.out.println(Error in CompletableFuture: throwable.getMessage());}});// 主线程继续执行其他操作System.out.println(Main thread continues...);// 等待所有操作完成CompletableFuture.allOf(firstOperation, secondOperation, thirdOperation).join();}
}Reactor优化Callback回调地狱
public class ReactorOptimizedExample {public static void main(String[] args) {Mono.fromCallable(() - {// 模拟异步操作Thread.sleep(2000);return First operation completed;}).subscribeOn(Schedulers.boundedElastic()).flatMap(result - {System.out.println(First Reactor: result);return Mono.fromCallable(() - {// 模拟异步操作Thread.sleep(2000);return Second operation completed with input: result;}).subscribeOn(Schedulers.boundedElastic());}).flatMap(result - {System.out.println(Second Reactor: result);return Mono.fromCallable(() - {// 模拟异步操作Thread.sleep(2000);return Third operation completed with input: result;}).subscribeOn(Schedulers.boundedElastic());}).doOnSuccess(result - System.out.println(Third Reactor: result)).doOnError(error - System.out.println(Error in Reactor: error.getMessage())).block(); // 阻塞等待操作完成// 主线程继续执行其他操作System.out.println(Main thread continues...);}
}学习打卡Java学习笔记-day06-响应式编程Reactor优化Callback回调地狱
