WebFlux批量调用外部接口：统一处理响应状态与结果

核心思路是用exchangeToMono替代retrieve——它能获取完整的ClientResponse对象，不会像retrieve那样自动把非2xx状态码转为异常，让我们可以自主处理所有状态码的响应，统一返回包含状态、数据或错误信息的结果。

第一步：定义统一响应封装类 #

先创建一个用来承载成功/失败信息的类，方便后续统一返回格式：

import org.springframework.http.HttpStatus;

public class ApiResponse<T> {
    private HttpStatus statusCode;
    private T data;
    private String errorMessage;

    // 成功场景构造器
    public ApiResponse(HttpStatus statusCode, T data) {
        this.statusCode = statusCode;
        this.data = data;
        this.errorMessage = null;
    }

    // 失败场景构造器
    public ApiResponse(HttpStatus statusCode, String errorMessage) {
        this.statusCode = statusCode;
        this.data = null;
        this.errorMessage = errorMessage;
    }

    // 按需添加getter、setter方法
}

第二步：修改原方法的WebClient调用逻辑 #

替换原有的retrieve为exchangeToMono，实现全状态码的自定义处理：

import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.ClientResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public Flux<ApiResponse<Entity>> getRatesFlux(String hash, String apiUrl) {
    return RatesReactiveRepository.findAll()
            .flatMap(rate -> {
                log.info("{} rate read from MongoDb: {}", logHash, rate);
                String url = apiUrl + "/" + rate.getRateName();
                log.debug("{}, START {} URI: {}",
                        logHash,
                        "productsController.ratesServiceImpl.getratesFlux",
                        url);

                return webClient.post()
                        .uri(url)
                        .headers(httpHeaders -> httpHeaders.set("hash", hash))
                        .accept(MediaType.APPLICATION_JSON)
                        // 用exchangeToMono获取完整响应，自主处理状态码
                        .exchangeToMono(clientResponse -> {
                            HttpStatus status = clientResponse.statusCode();
                            // 处理200 OK的成功场景
                            if (status.is2xxSuccessful() && status == HttpStatus.OK) {
                                return clientResponse.bodyToMono(Entity.class)
                                        .map(entity -> new ApiResponse<>(status, entity));
                            }
                            // 处理所有非200的状态（含其他2xx、4xx、5xx）
                            return clientResponse.bodyToMono(String.class)
                                    // 响应体为空时设置默认错误信息
                                    .defaultIfEmpty("No error message returned")
                                    .map(errorMsg -> new ApiResponse<>(status, errorMsg));
                        })
                        // 捕捉调用过程中的异常（如网络错误、超时），统一封装为错误结果
                        .onErrorResume(e -> Mono.just(new ApiResponse<>(HttpStatus.INTERNAL_SERVER_ERROR, e.getMessage())));
            });
}

关键逻辑说明 #

• exchangeToMono：核心是它不会自动抛出异常，所有状态码的响应都会进入自定义处理逻辑，完全掌控结果返回。
• 状态分支处理：明确区分200 OK和其他状态码，分别封装成功数据或错误信息。
• 异常兜底：用onErrorResume捕捉网络类异常，避免单个请求失败导致整个Flux中断。

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