继基础篇之后，本篇文章将深入探讨Spring WebFlux的进阶特性，帮助开发者在构建响应式应用时能够更好地管理复杂场景和提升性能。
一、进阶特性
错误处理策略
流程图：
响应式流中的错误处理是至关重要的。Spring WebFlux提供了多种错误处理机制，如onErrorReturn, onErrorResume, 和onErrorMap。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class ErrorHandlingExample {
public static void main(String[] args) {
// 创建一个包含错误的Flux序列
Flux<String> fluxWithError = Flux.just("A", "B", "C")
// 在序列中引入一个错误
.concatWith(Mono.error(new RuntimeException("Unexpected Error")))
// 使用onErrorReturn操作符来处理错误，返回一个默认值
.onErrorReturn("Default");
// 订阅并处理可能包含错误的Flux
fluxWithError.subscribe(
value -> System.out.println("Received: " + value),
error -> System.out.println("Error: " + error.getMessage()),
() -> System.out.println("Completed successfully")
);
// 创建另一个包含错误的Flux序列
Flux<String> fluxWithResume = Flux.just("1", "2", "error", "3")
// 使用flatMap来处理每个元素
.flatMap(value -> {
if ("error".equals(value)) {
// 当遇到"error"字符串时，抛出异常
return Mono.error(new IllegalStateException("Error occurred"));
} else {
// 其他情况下，正常返回值
return Mono.just(value);
}
})
// 使用onErrorResume来提供一个替代的Flux
.onErrorResume(e -> {
if (e instanceof IllegalStateException) {
// 如果错误是IllegalStateException，返回一个包含替代值的Flux
return Flux.just("Recovered");
} else {
// 其他类型的错误，重新抛出
return Flux.error(e);
}
});
// 订阅并处理可能包含错误的Flux
fluxWithResume.subscribe(
value -> System.out.println("Received: " + value),
error -> System.out.println("Error: " + error.getMessage()),
() -> System.out.println("Completed successfully")
);
// 创建一个包含错误的Mono序列
Mono<String> monoWithErrorMap = Mono.just("test")
// 模拟一个操作，该操作可能会抛出异常
.map(value -> {
if ("test".equals(value)) {
throw new RuntimeException("Test error");
}
return value;
})
// 使用onErrorMap来转换错误
.onErrorMap(original -> new CustomException("Custom error message", original));
// 订阅并处理可能包含错误的Mono
monoWithErrorMap.subscribe(
value -> System.out.println("Received: " + value),
error -> {
if (error instanceof CustomException) {
System.out.println("Custom Error: " + error.getMessage());
} else {
System.out.println("Error: " + error.getMessage());
}
},
() -> System.out.println("Completed successfully")
);
}
// 自定义异常类
static class CustomException extends RuntimeException {
public CustomException(String message, Throwable cause) {
super(message, cause);
}
}
}在这个示例中，我们创建了三个不同的响应式流，每个流都以不同的方式处理错误：
fluxWithError 使用 onErrorReturn 来返回一个默认值，当流中发生错误时。
fluxWithResume 使用 onErrorResume 来提供一个替代的流，这个替代的流会在原始流发生特定类型错误时发出一个恢复值。
monoWithErrorMap 使用 onErrorMap 来转换错误，将原始错误映射为一个自定义异常。
每个流都通过subscribe方法进行订阅，并提供了对应的处理逻辑。这些示例展示了如何在响应式流中优雅地处理错误情况。
调度器和线程模型
Spring WebFlux允许你通过调度器（Schedulers）来控制执行操作的线程，这对于管理异步任务的执行非常重要。
调度器定义了响应式流操作执行的上下文，即在哪个线程或线程池上执行。
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
public class SchedulerExample {
public static void main(String[] args) {
// 创建一个Flux序列
Flux<String> flux = Flux.just("A", "B", "C", "D")
// 指定一个调度器，该调度器使用一个弹性的线程池来运行流的处理
// 弹性线程池适用于I/O操作或其他阻塞操作，因为它可以根据需要创建新的线程
.subscribeOn(Schedulers.boundedElastic())
// map操作将在弹性线程池中执行，因为它位于subscribeOn之后
.map(value -> {
// 模拟耗时的操作，例如数据库调用
simulateExpensiveOperation(value);
return value.toLowerCase();
});
// 订阅并处理流
flux.subscribe(value -> {
// 打印线程信息和处理后的值
// 订阅者默认在发布者的线程中执行，除非使用publishOn指定其他调度器
System.out.println(Thread.currentThread().getName() + " -> " + value);
});
// 由于上面的操作是异步的，这里稍作等待，以便观察输出
// 在实际应用中，通常不需要这样做，因为Spring会管理流的生命周期
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private static void simulateExpensiveOperation(String value) {
try {
// 假设这是一个耗时的操作，例如远程服务调用或数据库操作
Thread.sleep(1000);
System.out.println("Processing " + value + " on thread " + Thread.currentThread().getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}在这个例子中，我们创建了一个包含四个字符串的Flux序列。我们使用subscribeOn(Schedulers.boundedElastic())来指定整个流的处理应该在一个弹性的线程池中执行。这个线程池适合执行阻塞操作，因为它可以根据需要动态地创建和销毁线程。
map操作符用于将字符串转换为小写，它会在弹性线程池中执行，因为它位于subscribeOn之后。最后，我们订阅了这个流，并打印出每个处理后的值以及执行该操作的线程信息。
请求合并
使用zip和combineLatest等操作符，可以将多个请求的结果合并起来，这在处理多个依赖的异步操作时非常有用。
请求合并是将多个异步操作的结果组合在一起，形成一个协调的响应。
zip操作符等待所有源Mono/Flux发出一个元素，并使用提供的函数将这些元素组合起来。combineLatest则在任何一个源发出元素时，使用最新发出的元素进行组合。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
public class RequestCombiningExample {
public static void main(String[] args) {
// 模拟两个异步数据源的请求
Mono<String> mono1 = Mono.just("Data from Service 1").delayElement(Duration.ofSeconds(1));
Mono<String> mono2 = Mono.just("Data from Service 2").delayElement(Duration.ofSeconds(2));
// 使用zip操作符合并两个Mono的结果
Mono<Tuple2<String, String>> combinedMono = Mono.zip(mono1, mono2);
// 订阅合并后的Mono并处理结果
combinedMono.subscribe(combinedResult -> {
// combinedResult是一个Tuple，包含两个Mono的结果
String service1Result = combinedResult.getT1();
String service2Result = combinedResult.getT2();
// 打印合并后的结果
System.out.println("Combined result: " + service1Result + " & " + service2Result);
});
// 由于上面的操作是异步的，这里稍作等待，以便观察输出
// 在实际应用中，通常不需要这样做，因为Spring会管理流的生命周期
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}在这个例子中，我们创建了两个Mono对象，分别模拟两个服务的异步响应。每个Mono都使用delayElement来模拟网络延迟。
我们使用Mono.zip操作符来合并这两个Mono的结果。zip操作符会等待所有参与合并的Mono都发出元素后，才会发出一个包含所有结果的Tuple。
然后我们订阅合并后的Mono并处理结果。在这里，我们从Tuple中提取每个服务的结果，并将它们打印出来。
最后，我们使用Thread.sleep(3000)来等待异步操作完成。这是为了演示目的，在实际的Spring WebFlux应用中，你通常不需要这样做，因为Spring的响应式框架会处理这些异步流的生命周期。
通过本篇文章，我们了解了Spring WebFlux的一些进阶特性，包括错误处理、线程模型、请求合并以及响应式流的转换和背压控制。这些特性为构建高效、健壮的响应式应用提供了更多的可能性。