Java - Reactor

Java Reactor 是一个用于构建响应式应用程序的库，核心组件包括 Flux（多元素流）和 Mono（单元素流），通过丰富的操作符实现异步、非阻塞的数据处理。 以下是关键概念和操作符的详细说明：

• Mono.just(T) / Flux.just(T...)创建包含单个或多个元素的发布者。

  Mono<String> mono = Mono.just("Hello");Flux<String> flux = Flux.just("Hello", "World");

• Mono.empty() / Flux.empty()创建不发射任何元素的发布者。

  Mono<Object> emptyMono = Mono.empty();

• Mono.error(Throwable) / Flux.error(Throwable)创建直接发射错误的发布者。

  Flux<Object> errorFlux = Flux.error(new RuntimeException("Error"));

• map(Function)同步转换每个元素。

  Flux.just(1, 2, 3).map(i -> i * 2); // 输出: 2, 4, 6

• flatMap(Function)异步转换元素为新的发布者并合并结果。

  Flux.just(1, 2, 3).flatMap(i -> Mono.just(i * 2)); // 输出: 2, 4, 6

• filter(Predicate)按条件过滤元素。

  Flux.just(1, 2, 3, 4).filter(i -> i > 2); // 输出: 3, 4

• switchIfEmpty(Mono)若原流为空，切换到备用流。

  Mono.<String>empty().switchIfEmpty(Mono.just("Default")); // 输出: "Default"

• concat(Flux) / concat(Mono)顺序连接多个流，前一个完成后处理下一个。

  Flux.concat(Flux.just(1, 2), Flux.just(3, 4)); // 输出: 1, 2, 3, 4

• zip(Flux, BiFunction)合并两个流的元素并应用函数。

  Flux.zip(Flux.just(1, 2), Flux.just("A", "B"), (i, s) -> i + s); // 输出: "1A", "2B"

• onErrorResume(Function)捕获错误并提供备用发布者。

  Mono.error(new Exception()).onErrorResume(e -> Mono.just("Recovered")); // 输出: "Recovered"

• doOnError(Consumer)仅执行错误副作用，不改变流。

  Flux.just(1, 2).concatWith(Flux.error(new RuntimeException())) .doOnError(System.err::println);

• doOnNext(Consumer)处理每个元素时触发操作。

  Flux.just(1, 2, 3).doOnNext(System.out::println); // 打印: 1, 2, 3

• doOnComplete(Runnable)流完成时执行操作。

  Flux.just(1, 2, 3).doOnComplete(() -> System.out.println("Done"));

• doFinally(Consumer)无论成功或失败，最终执行。

  Flux.just(1, 2).doFinally(type -> System.out.println("Signal: " + type));

• publishOn(Scheduler)指定后续操作的执行调度器。

  Flux.just(1, 2, 3).publishOn(Schedulers.boundedElastic());

• subscribeOn(Scheduler)指定订阅操作的调度器（影响整个流）。

  Flux.just(1, 2, 3).subscribeOn(Schedulers.parallel());

• 调度器类型

  Schedulers.parallel()固定线程池（默认 CPU 核心数），适合计算密集型任务。

  Schedulers.boundedElastic()弹性线程池（默认 10 * CPU 核心数），支持阻塞 I/O 操作。

• block()阻塞等待结果（仅限测试）。

  String result = Mono.just("Hello").block();

• timeout(Duration)设置超时限制。

  Flux.interval(Duration.ofMillis(100)).timeout(Duration.ofSeconds(1));

Reactor 通过 Flux 和 Mono 提供响应式编程能力，支持异步流处理、错误恢复、并发控制等。关键点包括：

• map vs flatMap：同步 vs 异步转换。
• concat vs zip：顺序连接 vs 并行组合。
• 调度器选择：parallel() 用于计算，boundedElastic() 用于阻塞 I/O。
• 避免阻塞：生产环境慎用 block()。

通过合理组合操作符和调度器，可以高效构建非阻塞、高并发的响应式应用。