Reactor流中的异常处理与资源清理：告别阻塞的finally

public Mono<Response> process(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);
}

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<Data> find(String id);
        Mono<Data> save(Data data);
    }

    // 模拟的外部API调用
    private Mono<Response> 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<Response> 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)) // 确保原始错误被重新传播
                )
            );
    }
}