Reactor Mono应用

使用案例

创建Mono

使用静态工厂方法创建Mono

import reactor.core.publisher.Mono;public class MonoExample {public static void main(String[] args) {// 创建一个包含值的MonoMono<String> monoWithValue = Mono.just("Hello, Reactor!");// 创建一个空的MonoMono<String> emptyMono = Mono.empty();// 创建一个包含错误的MonoMono<String> errorMono = Mono.error(new RuntimeException("Something went wrong"));}
}

从Callable、Supplier、CompletableFuture等创建Mono

import reactor.core.publisher.Mono;import java.util.concurrent.CompletableFuture;public class MonoExample {public static void main(String[] args) {// 从Callable创建MonoMono<String> callableMono = Mono.fromCallable(() -> "Hello from Callable");// 从Supplier创建MonoMono<String> supplierMono = Mono.fromSupplier(() -> "Hello from Supplier");// 从CompletableFuture创建MonoCompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello from CompletableFuture");Mono<String> futureMono = Mono.fromFuture(future);}
}

订阅Mono

订阅是开始数据流处理的关键步骤

import reactor.core.publisher.Mono;public class MonoExample {public static void main(String[] args) {Mono<String> mono = Mono.just("Hello, Reactor!");// 订阅并消费数据mono.subscribe(value -> System.out.println("Received: " + value), // onNexterror -> System.err.println("Error: " + error),   // onError() -> System.out.println("Completed")             // onComplete);}
}

操作Mono

转换和操作数据

import reactor.core.publisher.Mono;public class MonoExample {public static void main(String[] args) {Mono<String> mono = Mono.just("hello");// 转换数据Mono<String> transformedMono = mono.map(String::toUpperCase);// 链式操作transformedMono.flatMap(value -> Mono.just(value + " World")).subscribe(System.out::println);  // 输出: HELLO World}
}

异常处理

import reactor.core.publisher.Mono;public class MonoExample {public static void main(String[] args) {Mono<String> monoWithError = Mono.error(new RuntimeException("Original Error"));// 捕获并处理错误monoWithError.onErrorReturn("Fallback value").subscribe(System.out::println);  // 输出: Fallback value// 使用 onErrorResume 提供备用的 MonomonoWithError.onErrorResume(error -> {System.err.println("Error: " + error);return Mono.just("Recovered value");}).subscribe(System.out::println);  // 输出: Recovered value}
}

组合Mono

import reactor.core.publisher.Mono;public class MonoExample {public static void main(String[] args) {Mono<String> mono1 = Mono.just("Hello");Mono<String> mono2 = Mono.just("World");// 组合两个MonoMono<String> combinedMono = Mono.zip(mono1, mono2, (s1, s2) -> s1 + " " + s2);combinedMono.subscribe(System.out::println);  // 输出: Hello World}
}

调试和日志

使用日志功能可以帮助调试和监控数据流。

import reactor.core.publisher.Mono;
import reactor.util.Logger;
import reactor.util.Loggers;public class MonoExample {public static void main(String[] args) {Logger logger = Loggers.getLogger(MonoExample.class);Mono<String> mono = Mono.just("Hello, Reactor!").log();  // 默认日志mono.subscribe(System.out::println);}
}

调度器（Schedulers）

使用调度器来控制Mono的执行线程。

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;public class MonoExample {public static void main(String[] args) {Mono<String> mono = Mono.fromCallable(() -> {// 在独立的线程池中执行Thread.sleep(1000);return "Hello from another thread";});mono.subscribeOn(Schedulers.boundedElastic())  // 指定订阅时的调度器.publishOn(Schedulers.parallel())         // 指定发布时的调度器.subscribe(System.out::println);}
}

应用场景

异步计算

Mono可以用来表示和处理异步计算的结果。例如，当你需要从一个异步操作中获取一个值时，可以使用Mono。

Mono<String> asyncResult = Mono.fromCallable(() -> {// 模拟异步计算Thread.sleep(1000);return "Result";
});asyncResult.subscribe(result -> System.out.println("Received: " + result));

调用远程服务

在微服务架构中，调用远程服务（如REST API或gRPC）时，通常会返回一个单一的结果。这是Mono的一个典型应用场景。

WebClient webClient = WebClient.create("http://example.com");Mono<String> response = webClient.get().uri("/resource").retrieve().bodyToMono(String.class);response.subscribe(body -> System.out.println("Response: " + body));

数据库查询

Mono非常适合表示数据库查询返回的单个结果。例如，查询一个用户的信息

Mono<User> userMono = reactiveUserRepository.findById(userId);userMono.subscribe(user -> System.out.println("User: " + user));

事件驱动的处理

在事件驱动架构中，某些事件处理结果可能是单一的值。例如，处理某个事件并返回一个处理结果

Mono<EventResult> eventResultMono = processEvent(event);eventResultMono.subscribe(result -> System.out.println("Event processed: " + result));

错误处理

使用Mono可以优雅地处理异步操作中的错误。例如，如果某个操作可能会失败，可以返回一个错误的Mono并在订阅时处理错误

Mono<String> result = performOperation().onErrorReturn("Fallback value");result.subscribe(value -> System.out.println("Received: " + value),error -> System.err.println("Error: " + error)
);

延迟操作

Mono可以用于表示一个延迟操作，执行某些延迟逻辑

Mono<Long> delayMono = Mono.delay(Duration.ofSeconds(3));delayMono.subscribe(time -> System.out.println("Delayed for 3 seconds"));

条件逻辑

使用Mono可以在异步流中进行条件判断和逻辑处理。例如，根据某个条件返回不同的结果

Mono<String> conditionalMono = Mono.just("data").flatMap(data -> {if (data.equals("condition")) {return Mono.just("Condition met");} else {return Mono.just("Condition not met");}});conditionalMono.subscribe(System.out::println);

转换与映射

Mono可以用于对单个值进行转换或映射。例如，将一个值转换为另一个类型的值

Mono<String> originalMono = Mono.just("original");Mono<Integer> transformedMono = originalMono.map(String::length);transformedMono.subscribe(length -> System.out.println("Length: " + length));

资源管理

Mono可以用于在异步操作中管理资源，如文件或连接的打开和关闭

Mono.using(() -> new BufferedReader(new FileReader("data.txt")), reader -> Mono.fromCallable(() -> reader.readLine()), BufferedReader::close
).subscribe(line -> System.out.println("Read line: " + line));

组合多个异步操作

Mono可以用于组合多个异步操作，构建复杂的异步数据流

Mono<String> combinedMono = Mono.zip(Mono.just("Hello"),Mono.just("World"),(s1, s2) -> s1 + " " + s2
);combinedMono.subscribe(System.out::println);  // 输出: Hello World