Reactor流中finally阻塞操作整合及Spring WebFlux代码改造咨询

你的原同步代码核心逻辑是：查询已有数据→状态检查（不合法抛异常）→无数据则创建→调用外部API→无论API成败都保存数据→转换响应。转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);
   }
   
   return convertToResponse(existingData, response);
}

你尝试的改造代码（问题点标注） #

repository.find(request.getId())
    .flatMap(existingData -> {if (existingData.getState() != pending) {throw new RuntimeException("test");}})
    .switchIfEmpty(respository.save(convertToData(request)))
    .flaptMap(existingData -> hitAPI(existingData)) // 拼写错误：flaptMap → flatMap
    .flatMap(existingData -> convertToResponse(existingData)) // 原方法需要response参数，此处遗漏
    //? how to handle above error?

完整解决方案代码 #

public Mono<Response> process(Request request) {
    // 缓存数据查询/创建结果，避免重复执行
    Mono<Data> cachedData = repository.find(request.getId())
            // 状态检查：不合法则抛异常
            .flatMap(existing -> {
                if (existing.getState() != pending) {
                    return Mono.error(new RuntimeException("test"));
                }
                return Mono.just(existing);
            })
            // 无数据则创建并保存
            .switchIfEmpty(Mono.defer(() -> repository.save(convertToData(request))))
            .cache();

    return cachedData
            // 调用外部API，同时保留数据用于后续转换
            .flatMap(data -> 
                hitAPI(data)
                    // 捕获ServerException，日志后重新抛出
                    .onErrorResume(ServerException.class, e -> {
                        log.error("外部API调用失败", e);
                        return Mono.error(e);
                    })
                    // 转换为响应
                    .map(apiResp -> convertToResponse(data, apiResp))
            )
            // 对应原finally：无论成功/失败/取消，都执行阻塞保存
            .doFinally(signal -> 
                cachedData
                    .flatMap(data -> 
                        // 包装阻塞save为异步操作，指定弹性线程池
                        Mono.fromCallable(() -> repository.save(data))
                                .subscribeOn(Schedulers.boundedElastic())
                    )
                    // 手动订阅执行，doFinally不会自动触发订阅
                    .subscribe()
            );
}

关键细节说明 #

• 阻塞操作处理：原repository.save是阻塞方法，必须用Mono.fromCallable包裹，再通过subscribeOn(Schedulers.boundedElastic())分配到专门处理阻塞任务的线程池，绝对不能让阻塞操作占用WebFlux的IO线程，否则会导致服务响应能力急剧下降。
• finally语义实现：doFinally操作符会在流的生命周期结束时触发（不管是成功完成、抛出异常还是被取消），完全匹配原同步代码中finally块的行为。
• 错误处理对齐：用onErrorResume精准捕获ServerException，日志后重新抛出，和原代码的catch逻辑完全一致；状态检查不合法时用Mono.error()抛出异常，符合原代码的异常抛出逻辑。
• 避免重复操作：用cache()缓存数据查询/创建的结果，确保doFinally中的save操作不会重复执行查询或创建逻辑，提升性能。
• 参数传递修正：原convertToResponse需要existingData和API响应两个参数，改造时通过flatMap绑定两者，确保参数不遗漏。

内容的提问来源于stack exchange，提问作者doptimusprime