1. Reactor中的错误处理范式
在Project Reactor中，Mono和Flux通过错误信号（error signals）而非抛出异常来表示操作失败。因此，传统的try-catch机制在响应式链中是不适用的。为了处理这些错误信号，Reactor提供了一系列专用的操作符：
- doOnError(Consumer super Throwable> onError): 用于执行副作用操作，例如日志记录，它不会改变流的错误信号，错误会继续向下游传播。
- onErrorResume(Function super Throwable, ? extends Publisher extends T>> fallback): 当上游发出错误信号时，提供一个新的响应式流作为替代，下游将订阅这个新的流。这常用于错误恢复或提供默认值。
- onErrorMap(Function super Throwable, ? extends Throwable> errorMapper): 用于将一种错误类型转换为另一种错误类型，然后将转换后的错误向下游传播。
- onErrorContinue(...): 不推荐使用。 此操作符会吞噬错误并允许流继续处理后续元素，这通常会导致难以调试的逻辑错误和不一致的状态。应尽量避免使用。
2. 传统finally语义在Reactor中的实现
传统try-catch-finally中的finally块旨在无论代码块是否成功执行或抛出异常，都保证执行其中的逻辑，通常用于资源清理或状态更新。在Reactor中，由于其非阻塞和异步特性，实现finally语义需要更细致的考虑。
原始的阻塞式代码示例：
public Monoprocess(Request request) { // ... 前置逻辑 ... try { var response = hitAPI(existingData); } catch(ServerException serverException) { log.error(""); throw serverException; } finally { repository.save(existingData); // 阻塞操作 } return convertToResponse(existingData, response); }
问题在于，finally块中的repository.save(existingData)是一个阻塞操作，并且在响应式流中，我们需要确保无论成功还是失败，这个保存操作都能以非阻塞的响应式方式执行。
3. 响应式地重构“finally”逻辑
为了在Reactor中实现上述finally语义，我们需要将repository.save(existingData)这个操作集成到成功和失败的响应式流路径中。
以下是更符合Reactor范式的重构方案：
import reactor.core.publisher.Mono;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReactorService {
private static final Logger log = LoggerFactory.getLogger(ReactorService.class);
// 假设这些是响应式的接口
private ReactiveRepository repository; // 假设这是一个响应式仓库接口
private ReactiveApiService apiService; // 假设这是一个响应式API服务接口
// 构造函数或依赖注入
public ReactorService(ReactiveRepository repository, ReactiveApiService apiService) {
this.repository = repository;
this.apiService = apiService;
}
public Mono process(Request request) {
return repository.find(request.getId())
.flatMap(existingData -> {
// 检查状态，如果条件不满足，立即发出错误信号
if (existingData.getState() != State.PENDING) { // 假设State.PENDING是枚举值
return Mono.error(new RuntimeException("Data state is not pending."));
} else {
// 如果状态满足，继续保存或更新数据
return repository.save(convertToData(request)); // convertToData(request) 假设返回一个Mono
}
})
// 如果 find 没有找到数据，则 switchIfEmpty 会被触发，保存新数据
.switchIfEmpty(repository.save(convertToData(request))) // 假设 convertToData 返回 Mono
.flatMap(existingData -> Mono
// 调用外部API，使用 fromCallable 包装潜在的阻塞API调用
// 注意：理想情况下 hitAPI 应该返回 Mono/Flux
.fromCallable(() -> apiService.hitAPI(existingData))
.doOnError(ServerException.class, throwable -> log.error("API call failed: {}", throwable.getMessage(), throwable)) // 记录特定异常
// 错误处理路径：当 hitAPI 失败时
.onErrorResume(throwable ->
// 在错误发生时保存 existingData，然后重新发出原始错误
repository.save(existingData) // 假设 repository.save 返回 Mono
.then(Mono.error(throwable)) // 使用 then() 确保 save 完成后才发出错误
)
// 成功处理路径：当 hitAPI 成功时
.flatMap(response ->
// 在成功时保存 existingData，然后转换响应
repository.save(existingData) // 假设 repository.save 返回 Mono
.map(updatedExistingData -> convertToResponse(updatedExistingData, response)) // convertToResponse 假设返回 Response
)
);
}
// 辅助方法，根据实际业务逻辑定义
private Data convertToData(Request request) {
// 实际转换逻辑
return new Data(request.getId(), State.PENDING, "initial_data");
}
private Response convertToResponse(Data data, ApiResponse apiResponse) {
// 实际转换逻辑
return new Response(data.getId(), data.getState().name(), apiResponse.getStatus());
}
// 模拟接口和类
public enum State { PENDING, PROCESSED, FAILED }
public static class Request { String id; public Request(String id) { this.id = id; }}
public static class Response { String id; String status; String apiStatus; public Response(String id, String status, String apiStatus) { this.id = id; this.status = status; this.apiStatus = apiStatus; }}
public static class Data { String id; State state; String content; public Data(String id, State state, String content) { this.id = id; this.state = state; this.content = content; } public String getId() { return id; } public State getState() { return state; } public void setState(State state) { this.state = state; } public String getContent() { return content; }}
public static class ApiResponse { String status; public ApiResponse(String status) { this.status = status; }}
public static class ServerException extends RuntimeException { public ServerException(String message) { super(message); }}
public interface ReactiveRepository {
Mono find(String id);
Mono save(Data data);
}
public interface ReactiveApiService {
ApiResponse hitAPI(Data data) throws ServerException; // 模拟可能抛出ServerException的API
}
} 代码解析与注意事项：
-
替换阻塞操作：
- repository.find(request.getId()) 和 repository.save(...) 假设是返回Mono的响应式方法。
- hitAPI(existingData) 原本是同步阻塞的。在响应式流中，我们使用 Mono.fromCallable(() -> apiService.hitAPI(existingData)) 将其包装，使其在订阅时执行并在完成后发出结果。最佳实践是，hitAPI本身也应该是响应式的（例如返回Mono
），这样可以避免fromCallable带来的线程调度开销和潜在的阻塞风险。
-
错误信号处理：
- if (existingData.getState() != pending) { return Mono.error(new RuntimeException("test")); }：不再直接抛出异常，而是通过Mono.error()发出错误信号，这使得错误可以在响应式链中被捕获和处理。
- doOnError(ServerException.class, throwable -> log.error(...))：用于在ServerException发生时记录日志，这是一个副作用操作，不影响错误传播。
- onErrorResume(throwable -> repository.save(existingData).then(Mono.error(throwable)))：这是处理错误路径下“finally”逻辑的关键。当hitAPI操作失败时，onErrorResume会被触发。它首先执行repository.save(existingData)来更新状态，然后使用then(Mono.error(throwable))确保在save操作完成后，原始的错误信号被重新发出，以便下游可以继续处理这个错误。
-
成功路径下的“finally”逻辑：
- flatMap(response -> repository.save(existingData).map(updatedExistingData -> convertToResponse(updatedExistingData, response)))：这是处理成功路径下“finally”逻辑的关键。当hitAPI成功返回response后，我们接着执行repository.save(existingData)来更新状态。map操作符用于在save完成后，将更新后的existingData和response组合，转换为最终的Response。
避免重复逻辑的挑战： 如原始答案所述，finally中的repository.save(existingData)逻辑在响应式实现中被分解并复制到了成功 (flatMap) 和失败 (onErrorResume) 两个路径中。虽然这看起来是重复，但在Reactor中，这是确保无论结果如何都能执行特定操作的常见且必要的方式，因为流的控制流是基于成功信号或错误信号的。如果save操作是纯粹的资源释放且不影响流的后续数据或错误，doFinally可能是一个选项，但对于修改状态并可能影响后续流程的Mono操作，上述flatMap和onErrorResume的组合更为健壮。
总结
在Reactor中实现传统finally语义的关键在于将副作用操作（如保存数据）集成到响应式流的成功和错误路径中。通过doOnError进行日志记录，通过onErrorResume在错误时执行清理或恢复操作并重新发出错误，以及通过flatMap在成功时执行后续操作，我们可以构建出健壮且非阻塞的响应式流程。始终牢记，响应式编程的核心是避免阻塞，并利用操作符处理数据流和错误信号。
