1. 响应式编程中的挑战：阻塞与异常
在传统的命令式编程中，try-catch-finally结构是处理异常和确保资源清理的基石。然而，当我们将这种模式直接移植到Project Reactor等响应式框架中时，会遇到兼容性问题。响应式流是异步且非阻塞的，而finally块中的操作通常是同步且阻塞的。在一个响应式链中执行阻塞操作会严重损害其非阻塞特性，导致线程阻塞，影响系统吞吐量和响应速度。
此外，在Reactor中，不应直接抛出异常（throw new RuntimeException(...)），因为这会中断流的执行并跳过后续的响应式操作符。Reactor通过特殊的“错误信号”（error signal）来传播异常，这要求我们使用特定的操作符来处理这些信号。
2. Reactor错误处理的核心原则与操作符
Reactor中的Mono和Flux都内置了错误信号的概念。当流中发生错误时，它会发出一个错误信号并终止。为了捕获和处理这些错误，Reactor提供了一系列专用的操作符：
-
doOnError(Consumer
onError) : 用于执行带有副作用的操作，例如记录日志。它不会改变或恢复流，只是在错误发生时执行一个回调。 -
onErrorResume(Function
> fallback) : 用于在发生错误时提供一个备用流。如果上游发出错误信号，onErrorResume会订阅并切换到由其提供的新的Publisher（通常是Mono或Flux），从而实现错误恢复或降级。 -
onErrorMap(Function
errorMapper) : 用于将一种类型的错误转换为另一种类型。例如，将内部的IOException转换为业务相关的ServiceException。 -
onErrorContinue(BiConsumer
errorConsumer) : 强烈不推荐使用此操作符。 它的设计目的是在错误发生时跳过当前元素并继续处理后续元素，但这通常会导致难以理解的副作用和数据不一致性，因为它会“吞噬”错误信号。
3. 将finally逻辑融入响应式流
原先在finally块中执行的资源清理或状态保存操作，在响应式编程中需要被分解并整合到流的成功和错误路径中。这通常意味着需要在两个地方显式处理这些副作用：
- 成功路径: 当流正常完成时，执行相应的清理或保存操作。
- 错误路径: 当流因错误而终止时，执行相应的清理或保存操作。
让我们通过一个具体的例子来演示如何重构代码，使其符合Reactor的非阻塞和错误处理范式。
原始的命令式逻辑（存在阻塞问题）:
public Monoprocess(Request request) { var existingData = repository.find(request.getId()); // 假设是阻塞的 if (existingData != null) { if (existingData.getState() != pending) { throw new RuntimeException("test"); // 直接抛出异常 } } else { existingData = repository.save(convertToData(request)); // 假设是阻塞的 } try { var response = hitAPI(existingData); // 假设是阻塞的 } catch(ServerException serverException) { log.error(""); throw serverException; } finally { repository.save(existingData); // 阻塞的finally操作 } return convertToResponse(existingData, response); }
重构为响应式、非阻塞的Reactor风格：
假设repository是一个响应式仓库（返回Mono或Flux）。
import reactor.core.publisher.Mono;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReactiveProcessor {
private static final Logger log = LoggerFactory.getLogger(ReactiveProcessor.class);
private final ReactiveRepository repository; // 假设这是一个响应式仓库
public ReactiveProcessor(ReactiveRepository repository) {
this.repository = repository;
}
// 示例接口和类，实际应根据业务定义
interface Request { String getId(); }
interface Response {}
interface Data { String getState(); }
enum State { PENDING, COMPLETED } // 假设有PENDING状态
// 模拟的响应式仓库接口
interface ReactiveRepository {
Mono find(String id);
Mono save(Data data);
}
// 模拟的外部API调用
private Mono hitAPI(Data data) {
// 假设这是一个返回Mono的非阻塞API调用
// 如果是阻塞的，应使用 Mono.fromCallable 或 Mono.fromRunnable 包裹
return Mono.just(new Response() {}); // 示例
}
private Data convertToData(Request request) {
// 转换逻辑
return new Data() { @Override public String getState() { return State.PENDING.name(); } };
}
private Response convertToResponse(Data data, Response apiResponse) {
// 转换逻辑
return new Response() {};
}
public Mono process(Request request) {
return repository.find(request.getId())
// 1. 处理现有数据或创建新数据
.flatMap(existingData -> {
// 如果找到数据且状态不为PENDING，则发出错误信号
if (existingData.getState().equals(State.COMPLETED.name())) { // 假设COMPLETED是需要抛错的状态
return Mono.error(new RuntimeException("Data state is not pending."));
} else {
// 否则，返回现有数据
return Mono.just(existingData);
}
})
// 2. 如果find结果为空（switchIfEmpty），则保存新数据
.switchIfEmpty(Mono.defer(() -> repository.save(convertToData(request))))
// 3. 执行API调用并处理其结果及副作用
.flatMap(existingData ->
Mono.fromCallable(() -> { // 使用fromCallable包装可能阻塞的hitAPI（尽管这里假设hitAPI是响应式的）
// 实际业务中，hitAPI通常返回Mono，无需fromCallable
return hitAPI(existingData).block(); // 示例：模拟阻塞调用并立即阻塞，实际应避免
})
.flatMap(apiResponse -> {
// 成功路径：保存数据，然后转换为响应
return repository.save(existingData) // 模拟finally中的保存操作 (成功时)
.map(updatedData -> convertToResponse(updatedData, apiResponse));
})
// 4. 错误处理：记录日志并执行finally逻辑
.doOnError(ServerException.class, throwable -> log.error("API call failed: {}", throwable.getMessage(), throwable))
.onErrorResume(throwable ->
// 错误路径：保存数据，然后重新发出原始错误
repository.save(existingData) // 模拟finally中的保存操作 (错误时)
.then(Mono.error(throwable)) // 确保原始错误被重新传播
)
);
}
} 代码解析：
- 避免直接抛出异常: if (existingData.getState().equals(State.COMPLETED.name())) { return Mono.error(new RuntimeException("...")); } 代替了 throw new RuntimeException(...)。这是Reactor中传播错误信号的正确方式。
- switchIfEmpty: 用于处理repository.find返回空Mono的情况，此时会订阅repository.save(convertToData(request))来创建并保存新数据。Mono.defer确保save操作仅在需要时才被订阅。
- flatMap 链式调用: 整个流程通过flatMap连接起来，确保每个步骤都在上一步完成后异步执行。
- Mono.fromCallable: 尽管hitAPI在示例中被假设为返回Mono，但如果它确实是阻塞的，Mono.fromCallable是一个安全地将其封装进响应式流的方法。它会在一个单独的线程上执行提供的Callable，并将其结果包装成Mono。最佳实践是确保所有外部依赖（如API调用、数据库操作）本身就是响应式的。
- doOnError: doOnError(ServerException.class, throwable -> log.error(...)) 用于在ServerException发生时执行日志记录等副作用，而不会中断或改变错误流。
-
onErrorResume 中的 finally 逻辑:
- repository.save(existingData)：这是在错误发生时执行的“finally”逻辑。
- .then(Mono.error(throwable)): 在保存操作完成后，通过then操作符确保原始的错误信号被重新发出，以便下游操作符或订阅者能够继续处理该错误。
-
成功路径中的 finally 逻辑:
- repository.save(existingData).map(updatedData -> convertToResponse(updatedData, apiResponse)): 在API调用成功后，同样执行repository.save(existingData)，这是成功情况下的“finally”逻辑。然后，将更新后的数据和API响应转换为最终的Response。
4. 注意事项与总结
- 拥抱错误信号: 在Reactor中，将错误视为流的一部分，并使用Mono.error()或Flux.error()来发出错误信号，而不是传统的throw语句。
- 避免阻塞: 确保所有操作（包括数据库访问、外部API调用）都是非阻塞的。如果必须集成阻塞代码，使用Scheduler和Mono.fromCallable/Flux.fromIterable等操作符将其隔离到单独的线程池中。
- finally逻辑的分解: finally块中的逻辑需要被分解到响应式流的成功和错误路径中。doOnSuccess、doOnError、doFinally（对于无条件清理）以及在flatMap和onErrorResume中链式调用副作用操作是常见的方法。
- 副作用管理: doOn...系列操作符非常适合执行不影响流数据或错误的副作用（如日志记录、指标收集）。对于需要改变流行为或恢复的副作用，应使用flatMap、onErrorResume等操作符。
- 响应式仓库: 上述示例假设repository是一个响应式仓库。如果使用的是阻塞式JPA仓库，则需要使用Scheduler将其操作包装起来，例如Mono.fromCallable(() -> repository.save(data)).subscribeOn(Schedulers.boundedElastic())。
通过遵循这些原则和使用正确的Reactor操作符，我们可以构建出高效、健壮且完全非阻塞的响应式应用程序，优雅地处理异常和管理资源。
