如何在Spring WebFlux的exchange()方法外部获取ErrorBody填充POJO

问题描述 #

我在使用Spring WebFlux时遇到问题，需要将字符串格式的错误响应体填充到方法作用域之外声明的POJO（ExtraScope）中，现有代码无法正确实现该需求，具体代码如下：

ExtraScope extraScope = new ExtraScope();

Map<String, Object> vars = new HashMap<>();
vars.put("instrumentId", instrument);

RequestHeadersSpec<?> request = mdiWebclient
    .get()
    .uri(mdiEndpointsGetQuotations, vars)
    .header(HttpHeaders.AUTHORIZATION, token)
    .accept(MediaType.APPLICATION_JSON);

InstrumentDetailsDto block = request.exchangeToMono(r -> {
  if (r.statusCode() != HttpStatus.OK) {
      r.bodyToMono(String.class).subscribe(errorBody -> {
          logger.debug("errorBody: [{}] ", errorBody);
          extraScope.setBody(errorBody);
      });
      return Mono.empty();
    }
  return r.bodyToMono(InstrumentDetailsDto.class);
})
.block(Duration.ofMinutes(5));

System.out.println("content of ExtraScope pojo: " + extraScope.getBody());

问题分析 #

现有代码的核心问题是：在exchangeToMono中对r.bodyToMono(String.class)调用了subscribe()，这会触发一个独立的异步订阅，而主线程调用block()后会直接执行后续的System.out.println，此时异步回调可能还未完成，导致extraScope的body属性未被正确赋值。此外，这种在响应式流内部嵌套subscribe的写法违反了WebFlux响应式编程规范，易引发线程安全和时序问题。

正确实现方式 #

需要将错误响应体的处理整合到响应式流生命周期中，确保block()完成前赋值操作已执行。可以通过doOnNext和then操作符串联流程，具体代码如下：

ExtraScope extraScope = new ExtraScope();

Map<String, Object> vars = new HashMap<>();
vars.put("instrumentId", instrument);

RequestHeadersSpec<?> request = mdiWebclient
    .get()
    .uri(mdiEndpointsGetQuotations, vars)
    .header(HttpHeaders.AUTHORIZATION, token)
    .accept(MediaType.APPLICATION_JSON);

InstrumentDetailsDto block = request.exchangeToMono(r -> {
    if (!r.statusCode().is2xxSuccessful()) {
        // 将错误体获取与赋值整合到流中，确保block等待操作完成
        return r.bodyToMono(String.class)
                .doOnNext(errorBody -> {
                    logger.debug("errorBody: [{}] ", errorBody);
                    extraScope.setBody(errorBody);
                })
                .then(Mono.empty()); // 处理完错误体后返回空Mono
    }
    return r.bodyToMono(InstrumentDetailsDto.class);
})
.block(Duration.ofMinutes(5));

System.out.println("content of ExtraScope pojo: " + extraScope.getBody());

关键改进点 #

• 移除嵌套的subscribe()，改用doOnNext操作符在获取错误体时执行赋值，确保操作处于响应式流生命周期内。
• 使用then(Mono.empty())串联操作，保证错误体处理完成后再返回空Mono，确保block()等待整个流程结束。
• 将状态码判断改为is2xxSuccessful()，覆盖所有成功HTTP状态码，比单纯判断!= HttpStatus.OK更严谨。

额外优化建议 #

如果业务场景允许，更推荐将错误信息与结果纳入响应式流中，避免依赖外部可变对象（降低线程安全风险），示例如下：

Map<String, Object> vars = new HashMap<>();
vars.put("instrumentId", instrument);

RequestHeadersSpec<?> request = mdiWebclient
    .get()
    .uri(mdiEndpointsGetQuotations, vars)
    .header(HttpHeaders.AUTHORIZATION, token)
    .accept(MediaType.APPLICATION_JSON);

// 使用Tuple2同时返回结果和错误信息，或自定义包含两者的DTO
Tuple2<InstrumentDetailsDto, String> result = request.exchangeToMono(r -> {
    if (!r.statusCode().is2xxSuccessful()) {
        return r.bodyToMono(String.class)
                .map(errorBody -> Tuples.of(null, errorBody));
    }
    return r.bodyToMono(InstrumentDetailsDto.class)
            .map(dto -> Tuples.of(dto, null));
})
.block(Duration.ofMinutes(5));

InstrumentDetailsDto block = result.getT1();
String errorBody = result.getT2();
System.out.println("errorBody: " + errorBody);

这种方式无需外部可变对象，更符合响应式编程设计思想，避免潜在并发问题。

内容的提问来源于stack exchange，提问作者it.drive.view