南通高端网站设计建设,青岛新网站设计公司,泗水县城乡建设局网站,梦幻西游官网Reactor 是一个基于响应式编程的库#xff0c;主要用于构建异步和事件驱动的应用程序。Reactor 提供了丰富的 API#xff0c;包括创建、转换、过滤、组合等操作符#xff0c;用于处理异步数据流。以下是一些 Reactor 的主要 API 示例#xff1a; pom依赖 dependencyMan… Reactor 是一个基于响应式编程的库主要用于构建异步和事件驱动的应用程序。Reactor 提供了丰富的 API包括创建、转换、过滤、组合等操作符用于处理异步数据流。以下是一些 Reactor 的主要 API 示例 pom依赖 dependencyManagementdependenciesdependencygroupIdio.projectreactor/groupIdartifactIdreactor-bom/artifactIdversion2023.0.0/versiontypepom/typescopeimport/scope/dependency/dependencies/dependencyManagementdependenciesdependencygroupIdio.projectreactor/groupIdartifactIdreactor-core/artifactId/dependencydependencygroupIdio.projectreactor/groupIdartifactIdreactor-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.junit.jupiter/groupIdartifactIdjunit-jupiter/artifactIdversion5.7.2/versionscopetest/scope/dependency/dependencies61. 使用 Reactor 的 then 方法进行后续操作
then 方法用于在当前数据流完成后执行后续操作。
import reactor.core.publisher.Flux;public class ReactorThenExample {public static void main(String[] args) {FluxInteger source Flux.just(1, 2, 3);// 在当前数据流完成后执行后续操作source.then(Mono.fromRunnable(() - System.out.println(Done))).subscribe();}
}62. 使用 Reactor 的 publishOn 方法进行线程切换
publishOn 方法用于切换数据流的发布线程从而改变元素处理的线程。
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;public class ReactorPublishOnExample {public static void main(String[] args) {FluxInteger source Flux.range(1, 3);// 将数据流的发布线程切换到另一个线程池source.publishOn(Schedulers.elastic()).map(value - value * 2).subscribe(System.out::println);}
}63. 使用 Reactor 的 subscribeOn 方法进行订阅线程切换
subscribeOn 方法用于切换数据流的订阅线程影响整个数据流的执行线程。
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;public class ReactorSubscribeOnExample {public static void main(String[] args) throws InterruptedException {FluxInteger source Flux.range(1, 3).log();// 将数据流的订阅线程切换到另一个线程池 另一个线程parallel-1source.subscribeOn(Schedulers.parallel()).map(value - value * 2).subscribe(System.out::println);Thread.sleep(23333);}
}64. 使用 Reactor 的 delayElements 方法进行元素延迟
delayElements 方法用于延迟数据流中元素的发送。
import reactor.core.publisher.Flux;
import java.time.Duration;public class ReactorDelayElementsExample {public static void main(String[] args) throws InterruptedException {FluxInteger source Flux.range(1, 3);// 延迟每个元素的发送source.delayElements(Duration.ofSeconds(1)).subscribe(System.out::println);Thread.sleep(23333);}
}65. 使用 Reactor 的 concatWith 方法进行数据流连接
concatWith 方法用于将两个数据流连接在一起保持顺序。
import reactor.core.publisher.Flux;public class ReactorConcatWithExample {public static void main(String[] args) {FluxInteger source1 Flux.just(1, 2, 3);FluxInteger source2 Flux.just(4, 5, 6);// 将两个数据流连接在一起保持顺序source1.concatWith(source2).subscribe(System.out::println);}
}66. 使用 Reactor 的 merge 方法进行多数据流合并
merge 方法用于将多个数据流合并成一个数据流并发执行。
import reactor.core.publisher.Flux;public class ReactorMergeExample {public static void main(String[] args) {FluxInteger source1 Flux.just(1, 2, 3);FluxInteger source2 Flux.just(4, 5, 6);// 将两个数据流合并成一个数据流FluxInteger mergedFlux Flux.merge(source1, source2);mergedFlux.subscribe(System.out::println);}
}67. concatWith和merge的比较
执行顺序 concatWith 这个方法会按照合并的顺序执行 Flux。它会等待第一个 Flux 完成包括完成信号或错误信号然后再开始下一个 Flux。merge 这个方法会并发执行所有的 Flux它不会等待前一个 Flux 完成。因此元素的顺序可能是交错的。 参数类型 concatWith 它接受一个单独的 Flux 作为参数将这个 Flux 追加到当前 Flux 的末尾。merge 它接受可变参数可以传入多个 Flux并同时合并它们。
public class FluxConcatWithMergeExample {public static void main(String[] args) throws InterruptedException {FluxInteger flux1 Flux.just(1, 2, 3).delayElements(Duration.ofMillis(100));FluxInteger flux2 Flux.just(4, 5, 6).delayElements(Duration.ofMillis(50));FluxInteger flux3 Flux.just(7, 8, 9).delayElements(Duration.ofMillis(75));// 使用 concatWith 方法按顺序执行flux1.concatWith(flux2).concatWith(flux3).subscribe(v -{System.out.println(concatWith v);});// 使用 merge 方法并发执行Flux.merge(flux1, flux2, flux3).subscribe(v -{System.out.println(merge v);});Thread.sleep(22333);}
}68. 使用 Reactor 的 mergeSequential 方法进行多数据流合并
mergeSequential 方法用于按顺序合并多个数据流保持各个数据流的元素顺序。
import reactor.core.publisher.Flux;public class ReactorMergeSequentialExample {public static void main(String[] args) {FluxInteger source1 Flux.just(1, 2, 3);FluxInteger source2 Flux.just(4, 5, 6);// 按顺序合并两个数据流FluxInteger mergedFlux Flux.mergeSequential(source1, source2);mergedFlux.subscribe(System.out::println);}
}69. 使用 Reactor 的 combineLatest 方法进行多数据流合并
combineLatest 方法用于合并多个数据流的最新元素。
import reactor.core.publisher.Flux;public class ReactorCombineLatestExample {public static void main(String[] args) {FluxInteger source1 Flux.just(1, 2, 3);FluxInteger source2 Flux.just(4, 5, 6);// 合并两个数据流的最新元素FluxInteger combinedFlux Flux.combineLatest(source1, source2, (a, b) - a b);combinedFlux.subscribe(System.out::println);}
}71. 使用 Reactor 的 doOnNext 方法进行每个元素的附加操作
doOnNext 方法用于在每个元素发出时执行附加操作例如日志记录、统计等。
import reactor.core.publisher.Flux;public class ReactorDoOnNextExample {public static void main(String[] args) {FluxInteger source Flux.range(1, 3);// 在每个元素发出时执行附加操作source.doOnNext(value - System.out.println(Processing: value)).subscribe(System.out::println);}
}79. 使用 Reactor 的 fromCallable 方法创建带有返回值的 Mono
fromCallable 方法用于创建一个 Mono其值由提供的 Callable 对象返回。
import reactor.core.publisher.Mono;import java.util.concurrent.Callable;public class ReactorFromCallableExample {public static void main(String[] args) {// 创建带有返回值的 MonoMonoString resultMono Mono.fromCallable(() - {// 执行一些计算return Result;});resultMono.subscribe(System.out::println);}
}80. 使用 Reactor 的 using 方法进行资源管理
using 方法用于在数据流的生命周期内管理资源例如打开和关闭文件、网络连接等。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;public class ReactorUsingExample {public static void main(String[] args) {// 使用 using 方法管理资源FluxString resultFlux Flux.using(() - getResource(), // 打开资源resource - getData(resource), // 使用资源获取数据流resource - releaseResource(resource) // 关闭资源);resultFlux.subscribe(System.out::println);}private static MonoString getResource() {System.out.println(Opening resource);return Mono.just(Resource);}private static FluxString getData(Mono resource) {System.out.println(Getting data from resource: resource);return Flux.just(Data1, Data2, Data3);}private static MonoVoid releaseResource(Mono resource) {System.out.println(Releasing resource: resource);return Mono.empty();}
}82. 使用 Reactor 的 scan 方法进行累积操作
scan 方法用于对数据流中的元素进行累积操作并生成一个新的数据流。
import reactor.core.publisher.Flux;public class ReactorScanExample {public static void main(String[] args) {FluxInteger source Flux.just(1, 2, 3, 4, 5);// 对数据流中的元素进行累积操作source.scan(0, (acc, value) - acc value).subscribe(System.out::println);}
}83. 使用 Reactor 的 takeWhile 方法进行条件性的元素获取
takeWhile 方法用于根据指定的条件获取数据流中的元素直到条件不满足。
import reactor.core.publisher.Flux;public class ReactorTakeWhileExample {public static void main(String[] args) {FluxInteger source Flux.just(1, 2, 3, 4, 5);// 根据条件获取元素直到条件不满足source.takeWhile(value - value 4).subscribe(System.out::println);}
}84. 使用 Reactor 的 thenMany 方法进行串联操作
thenMany 方法用于在当前数据流完成后执行另一个数据流将它们串联起来。
import reactor.core.publisher.Flux;public class ReactorThenManyExample {public static void main(String[] args) {FluxInteger source1 Flux.just(1, 2, 3);FluxInteger source2 Flux.just(4, 5, 6);// 在当前数据流完成后执行另一个数据流source1.thenMany(source2).subscribe(System.out::println);}
}85. 使用 Reactor 的 ignoreElements 方法忽略所有元素
ignoreElements 方法用于忽略数据流中的所有元素只关注完成信号或错误信号。
import reactor.core.publisher.Flux;public class ReactorIgnoreElementsExample {public static void main(String[] args) {FluxInteger source Flux.just(1, 2, 3);// 忽略所有元素只关注完成信号source.ignoreElements().doOnTerminate(() - System.out.println(Completed)).subscribe();}
}在 Reactor 中Sink 是一个用于手动推送元素signals到 Subscriber 的接口。它允许你在创建 Flux 或 Mono 的过程中手动控制元素的生成。Reactor 提供了两种 SinkFluxSink 用于创建 FluxMonoSink 用于创建 Mono。
98. 使用 FluxSink 发送元素和完成信号
import reactor.core.publisher.Flux;public class FluxSinkExample {public static void main(String[] args) {Flux.create(fluxSink - {for (int i 0; i 5; i) {fluxSink.next(i); // 发送元素}fluxSink.complete(); // 发送完成信号}).subscribe(System.out::println);}
}99. 使用 FluxSink 发送元素和错误信号
import reactor.core.publisher.Flux;public class FluxSinkErrorExample {public static void main(String[] args) {Flux.create(fluxSink - {for (int i 0; i 5; i) {fluxSink.next(i); // 发送元素}fluxSink.error(new RuntimeException(Simulated error)); // 发送错误信号}).subscribe(System.out::println,error - System.err.println(Error: error.getMessage()));}
}100. 使用 MonoSink 发送元素和完成信号
import reactor.core.publisher.Mono;public class MonoSinkExample {public static void main(String[] args) {Mono.create(monoSink - {monoSink.success(Hello, Mono!); // 发送元素}).subscribe(System.out::println);}
}101. 使用 MonoSink 发送错误信号
import reactor.core.publisher.Mono;public class MonoSinkErrorExample {public static void main(String[] args) {Mono.create(monoSink - {monoSink.error(new RuntimeException(Simulated error)); // 发送错误信号}).subscribe(System.out::println,error - System.err.println(Error: error.getMessage()));}
}102. 使用 FluxSink 进行背压控制
在 Reactor 中FluxSink 也提供了一些方法用于实现背压控制以避免在高速生产者和低速消费者之间的元素溢出。
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;public class FluxSinkBackpressureExample {public static void main(String[] args) {Flux.create(fluxSink - {for (int i 0; i 1000; i) {fluxSink.next(i);}}, FluxSink.OverflowStrategy.BUFFER) // 指定背压策略.onBackpressureBuffer(10, buffer - System.err.println(Buffer overflow! Discarding: buffer)).subscribe(value - {// 模拟慢速消费者try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(value);});}
}在上述例子中通过指定 FluxSink.OverflowStrategy.BUFFER 背压策略当消费者无法跟上生产者的速度时缓冲区将被用来存储元素。使用 onBackpressureBuffer 方法可以在溢出时执行自定义的操作。
103. 使用 FluxSink 进行手动请求
FluxSink 也提供了 request 方法允许消费者手动请求元素。
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;public class FluxSinkManualRequestExample {public static void main(String[] args) {Flux.create(fluxSink - {for (int i 0; i 100; i) {fluxSink.next(i);if (i % 10 0 fluxSink.requestedFromDownstream() 0) {// 当请求的元素达到 0 时等待下游再次请求while (fluxSink.requestedFromDownstream() 0) {// 等待下游请求try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}}}}fluxSink.complete();}).subscribe(System.out::println);}
}在这个例子中当消费者请求的元素达到 0 时生产者会等待下游再次请求。这种手动控制请求的方式可以更灵活地处理背压。
107. 使用 Reactor 的 Hooks 进行全局错误处理
Hooks 是 Reactor 提供的一组钩子可以用于全局错误处理捕获整个流的错误。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;public class ReactorHooksErrorHandlingExample {public static void main(String[] args) {// 使用 Hooks 进行全局错误处理Hooks.onOperatorError((error, reference) - {System.err.println(Global Error Handling: error.getMessage());return error;});FluxInteger source Flux.just(1, 2, 0, 4, 5);// 流中的错误将被全局处理source.map(x - 10 / x).subscribe(data - System.out.println(Received: data),error - System.err.println(Subscriber Error: error.getMessage()));}
}在这个例子中我们使用 Hooks.onOperatorError 来设置全局错误处理当流中发生错误时会调用全局错误处理的回调方法。这可以用于捕获整个流的错误而不是每个 subscribe 中单独处理。
109. 使用 Reactor 的 ConnectableFlux 进行热序列
ConnectableFlux 是 Reactor 提供的一种特殊类型的 Flux它允许在订阅之前预热开始生成元素并在多个订阅者之间共享相同的序列。
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;import java.time.Duration;public class ReactorConnectableFluxExample {public static void main(String[] args) {ConnectableFluxInteger connectableFlux Flux.range(1, 3).delayElements(Duration.ofSeconds(1)).publish(); // 将普通的 Flux 转换为 ConnectableFluxconnectableFlux.connect(); // 开始生成元素// 第一个订阅者connectableFlux.subscribe(data - System.out.println(Subscriber 1: data));try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}// 第二个订阅者共享相同的序列connectableFlux.subscribe(data - System.out.println(Subscriber 2: data));// 结果// Subscriber 1: 1// Subscriber 1: 2// Subscriber 2: 2// Subscriber 1: 3// Subscriber 2: 3}
}在这个例子中我们使用 publish 方法将普通的 Flux 转换为 ConnectableFlux通过 connect 方法开始生成元素。第一个订阅者在元素生成过程中订阅然后等待了 2 秒后第二个订阅者也开始订阅两者共享相同的序列。这种方式可以用于创建热序列使得订阅者能够共享相同的元素序列。
110. 使用 Reactor 的 Flux.defer 实现延迟订阅
Flux.defer 允许你在每次订阅时创建一个新的 Flux从而实现延迟订阅。这对于需要在每次订阅时执行一些逻辑的场景非常有用。
import reactor.core.publisher.Flux;public class ReactorFluxDeferExample {public static void main(String[] args) {FluxInteger deferredFlux Flux.defer(() - {// 在每次订阅时创建新的 FluxSystem.out.println(Creating new Flux);return Flux.just(1, 2, 3);});// 第一个订阅deferredFlux.subscribe(data - System.out.println(Subscriber 1: data));// 第二个订阅deferredFlux.subscribe(data - System.out.println(Subscriber 2: data));// 结果// Creating new Flux// Subscriber 1: 1// Subscriber 1: 2// Subscriber 1: 3// Creating new Flux// Subscriber 2: 1// Subscriber 2: 2// Subscriber 2: 3}
}在这个例子中Flux.defer 中的 lambda 表达式将在每次订阅时执行因此每个订阅都会创建一个新的 Flux。这对于那些需要在每次订阅时重新生成数据的情况非常有用。
119. 使用 Reactor 的 Flux.handle 处理元素和错误
Flux.handle 方法用于处理元素和错误通过提供一个 BiConsumer 处理每个元素并通过提供一个 BiConsumer 处理错误。
import reactor.core.publisher.Flux;public class ReactorFluxHandleExample {public static void main(String[] args) {FluxInteger source Flux.just(1, 2, 0, 4, 5);// 处理元素和错误FluxInteger handledFlux source.handle((value, sink) - {if (value ! 0) {sink.next(value); // 处理元素} else {sink.error(new RuntimeException(Cannot divide by zero)); // 处理错误}});handledFlux.subscribe(System.out::println,error - System.err.println(Error: error.getMessage()));}
}在这个例子中我们使用 Flux.handle 处理每个元素如果元素不为零则将其发送到下游如果元素为零则通过 sink.error 处理错误。这可以用于处理元素和错误的场景。
120. 使用 Reactor 的 Mono.handle 处理元素和错误
Mono.handle 方法与 Flux.handle 类似用于处理单个元素和错误。
import reactor.core.publisher.Mono;public class ReactorMonoHandleExample {public static void main(String[] args) {MonoInteger source Mono.just(10);// 处理元素和错误MonoInteger handledMono source.handle((value, sink) - {if (value 0) {sink.next(value); // 处理元素} else {sink.error(new RuntimeException(Invalid value)); // 处理错误}});handledMono.subscribe(System.out::println,error - System.err.println(Error: error.getMessage()));}
}在这个例子中我们使用 Mono.handle 处理单个元素如果元素为正数则发送到下游如果元素不为正数则通过 sink.error 处理错误。这可以用于处理单个元素和错误的场景。 太太太多了到此为止吧~~~~ 学习打卡day08响应式编程Reactor API大全下
