在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); // 调用外部API
   } catch(ServerException serverException) {
     log.error("");
     throw serverException; // API调用失败则抛异常
   } finally {
     repository.save(existingData); // 无论成功失败，都保存数据
   }

   return convertToResponse(existingData, response); // 转换响应
}

import reactor.core.publisher.Mono;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// 假设的依赖和实体
class Request { String getId() { return null; } }
class Response {}
class Data { Object getState() { return null; } } // 假设有getState方法
enum State { pending, completed } // 假设有pending状态
class ServerException extends RuntimeException {}

// 假设的Repository接口（返回Mono）
interface ReactiveRepository {
    Mono<Data> find(String id);
    Mono<Data> save(Data data);
}

public class ReactiveProcessService {

    private static final Logger log = LoggerFactory.getLogger(ReactiveProcessService.class);
    private final ReactiveRepository repository;

    public ReactiveProcessService(ReactiveRepository repository) {
        this.repository = repository;
    }

    private Data convertToData(Request request) { /* 转换逻辑 */ return new Data(); }
    private Response convertToResponse(Data data, Object response) { /* 转换逻辑 */ return new Response(); }
    private Object hitAPI(Data data) throws ServerException { /* 模拟外部API调用 */ return new Object(); }

    public Mono<Response> process(Request request) {
        return repository.find(request.getId())
                .flatMap(existingData -> {
                    // 如果找到现有数据
                    if (existingData.getState() != State.pending) {
                        // 如果状态不是pending，则发出错误信号
                        return Mono.error(new RuntimeException("Data state is not pending."));
                    } else {
                        // 如果状态是pending，则继续使用现有数据
                        return Mono.just(existingData);
                    }
                })
                .switchIfEmpty(Mono.defer(() -> repository.save(convertToData(request)))) // 如果未找到数据，则保存新数据
                .flatMap(existingData -> Mono
                        // 包装可能阻塞的API调用，使其在响应式流中执行
                        .fromCallable(() -> hitAPI(existingData))
                        // 捕获ServerException，记录日志，但不中断流（错误信号会继续传播）
                        .doOnError(ServerException.class, throwable -> log.error("API call failed: {}", throwable.getMessage(), throwable))
                        // 错误处理路径：如果API调用失败，先保存数据，再重新发出错误信号
                        .onErrorResume(throwable -> 
                            repository.save(existingData) // 执行“finally”逻辑：保存数据
                                .then(Mono.error(throwable)) // 然后重新发出原始错误信号
                        )
                        // 成功处理路径：如果API调用成功，先保存数据，再转换响应
                        .flatMap(apiResponse -> 
                            repository.save(existingData) // 执行“finally”逻辑：保存数据
                                .map(updatedExistingData -> convertToResponse(updatedExistingData, apiResponse))
                        )
                );
    }
}