Reactor流中“finally”语义的实现与阻塞操作的规避

public Mono<Response> process(Request request) {
   // ... 前置逻辑 ...
   try {
     var response = hitAPI(existingData);
   } catch(ServerException serverException) {
     log.error("");
     throw serverException;
   } finally {
     repository.save(existingData); // 阻塞操作
   }
   return convertToResponse(existingData, response);
}

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<Response> 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<Data>
                }
            })
            // 如果 find 没有找到数据，则 switchIfEmpty 会被触发，保存新数据
            .switchIfEmpty(repository.save(convertToData(request))) // 假设 convertToData 返回 Mono<Data>
            .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<Data>
                        .then(Mono.error(throwable)) // 使用 then() 确保 save 完成后才发出错误
                )
                // 成功处理路径：当 hitAPI 成功时
                .flatMap(response ->
                    // 在成功时保存 existingData，然后转换响应
                    repository.save(existingData) // 假设 repository.save 返回 Mono<Data>
                        .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<Data> find(String id);
        Mono<Data> save(Data data);
    }

    public interface ReactiveApiService {
        ApiResponse hitAPI(Data data) throws ServerException; // 模拟可能抛出ServerException的API
    }
}