Java 11中java.net.http.HttpClient多BodySubscriber响应解析异常处理问题

问题场景 #

使用JsonBodyHandler解析请求成功的响应体，同时定义ExceptionModel处理请求失败的场景，需要根据响应状态码将响应映射为预期模型或异常模型类。

现有代码实现 #

JsonBodyHandler类 #

private static class JsonBodyHandler<R> implements HttpResponse.BodyHandler<Supplier<R>> {

    private final Class<R> returnClz;

    public JsonBodyHandler(Class<R> returnClz) {
      this.returnClz = returnClz;
    }

    private static <R> HttpResponse.BodySubscriber<Supplier<R>> asJson(Class<R> returnClz) {
      HttpResponse.BodySubscriber<InputStream> upstream =
          HttpResponse.BodySubscribers.ofInputStream();

      return HttpResponse.BodySubscribers.mapping(
          upstream, inputStream -> toSupplierOfType(inputStream, returnClz));
    }

    private static <R> Supplier<R> toSupplierOfType(InputStream inputStream, Class<R> returnClz) {
      return () -> {
        try (InputStream stream = inputStream) {
          ObjectMapper objectMapper = new ObjectMapper();
          return objectMapper.readValue(stream, returnClz);
        } catch (IOException e) {
          throw new UncheckedIOException(e);
        }
      };
    }

    @Override
    public HttpResponse.BodySubscriber<Supplier<R>> apply(HttpResponse.ResponseInfo responseInfo) {
      return asJson(returnClz);
    }
}

请求发送方法 #

<B, R> R exchange(URI uri, String method, B body, Class<R> returnClz) {
    Builder httpRequestBuilder = HttpRequest.newBuilder().uri(uri);
    addHeaders(httpRequestBuilder);

    var httpRequest =
        "GET".equals(method)
            ? httpRequestBuilder.GET().build()
            : httpRequestBuilder.method(method, getBodyPublisher(body)).build();

    var httpClient = HttpClient.newHttpClient();
    JsonBodyHandler<R> bodyHandler = new JsonBodyHandler<>(returnClz);
    Supplier<R> responseSupplier = httpClient.send(httpRequest, bodyHandler).body();
    return responseSupplier.get();
}

尝试方案及编译错误 #

尝试根据状态码选择不同的BodySubscriber，但出现编译错误：

BodySubscriber<Supplier<R>> successBodySubscriber = JsonBodyHandler.asJson(returnClz);
BodySubscriber<Supplier<ExceptionModel>> failureBodySubscriber =
    JsonBodyHandler.asJson(ExceptionModel.class);
BodyHandler<Supplier> jsonBodyHandler =
    (rspInfo) -> rspInfo.statusCode() == 200 ? successBodySubscriber : failureBodySubscriber;

HttpResponse<Supplier> httpResponse = httpClient.send(httpRequest, jsonBodyHandler);
if (httpResponse.statusCode() != 200) {
  Supplier<ExceptionModel> responseSupplier = httpResponse.body();
  throw ClientServiceError.invalidResponse(responseSupplier.get());
}

Supplier<R> responseSupplier = httpResponse.body();
return responseSupplier.get();

编译错误信息：

Incompatible types. Found: 'java.net.http.HttpResponse.BodySubscriber<java.util.function.Supplier>', required: 'java.net.http.HttpResponse.BodySubscriber<java.util.function.Supplier>'
Incompatible types. Found: 'java.net.http.HttpResponse.BodySubscriber<java.util.function.Supplier<com.project.proxy.impl.ExceptionModel>>', required: 'java.net.http.HttpResponse.BodySubscriber<java.util.function.Supplier>'

注意：使用Java 11，无法升级版本。

可行解决方案 #

方案1：统一返回类型为Supplier<?> #

通过泛型通配符统一不同类型的Subscriber，解决编译类型不兼容问题：

// 统一类型为Supplier<?>，兼容两种解析结果
BodySubscriber<Supplier<?>> successBodySubscriber = JsonBodyHandler.asJson(returnClz);
BodySubscriber<Supplier<?>> failureBodySubscriber = JsonBodyHandler.asJson(ExceptionModel.class);

BodyHandler<Supplier<?>> jsonBodyHandler = rspInfo -> 
    rspInfo.statusCode() == 200 ? successBodySubscriber : failureBodySubscriber;

HttpResponse<Supplier<?>> httpResponse = httpClient.send(httpRequest, jsonBodyHandler);
if (httpResponse.statusCode() != 200) {
  // 基于状态码的强制转换是安全的
  Supplier<ExceptionModel> exceptionSupplier = (Supplier<ExceptionModel>) httpResponse.body();
  throw ClientServiceError.invalidResponse(exceptionSupplier.get());
}

// 强制转换为预期类型
Supplier<R> responseSupplier = (Supplier<R>) httpResponse.body();
return responseSupplier.get();

方案2：封装动态解析逻辑到Handler内部 #

扩展JsonBodyHandler，让它内部根据状态码自动切换解析类型：

private static class DynamicJsonBodyHandler<R> implements HttpResponse.BodyHandler<Supplier<?>> {

    private final Class<R> successClz;
    private final Class<ExceptionModel> failureClz;

    public DynamicJsonBodyHandler(Class<R> successClz) {
        this.successClz = successClz;
        this.failureClz = ExceptionModel.class;
    }

    private static <T> HttpResponse.BodySubscriber<Supplier<T>> asJson(Class<T> returnClz) {
        HttpResponse.BodySubscriber<InputStream> upstream = HttpResponse.BodySubscribers.ofInputStream();
        return HttpResponse.BodySubscribers.mapping(upstream, inputStream -> toSupplierOfType(inputStream, returnClz));
    }

    private static <T> Supplier<T> toSupplierOfType(InputStream inputStream, Class<T> returnClz) {
        return () -> {
            try (InputStream stream = inputStream) {
                ObjectMapper objectMapper = new ObjectMapper();
                return objectMapper.readValue(stream, returnClz);
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        };
    }

    @Override
    public HttpResponse.BodySubscriber<Supplier<?>> apply(HttpResponse.ResponseInfo responseInfo) {
        // 根据状态码选择解析类型
        if (responseInfo.statusCode() == 200) {
            return asJson(successClz);
        } else {
            return asJson(failureClz);
        }
    }
}

在请求方法中使用该动态Handler：

<B, R> R exchange(URI uri, String method, B body, Class<R> returnClz) {
    Builder httpRequestBuilder = HttpRequest.newBuilder().uri(uri);
    addHeaders(httpRequestBuilder);

    var httpRequest =
        "GET".equals(method)
            ? httpRequestBuilder.GET().build()
            : httpRequestBuilder.method(method, getBodyPublisher(body)).build();

    var httpClient = HttpClient.newHttpClient();
    DynamicJsonBodyHandler<R> bodyHandler = new DynamicJsonBodyHandler<>(returnClz);
    HttpResponse<Supplier<?>> httpResponse = httpClient.send(httpRequest, bodyHandler);
    
    if (httpResponse.statusCode() != 200) {
        Supplier<ExceptionModel> exceptionSupplier = (Supplier<ExceptionModel>) httpResponse.body();
        throw ClientServiceError.invalidResponse(exceptionSupplier.get());
    }

    Supplier<R> responseSupplier = (Supplier<R>) httpResponse.body();
    return responseSupplier.get();
}

方案说明 #

• 两种方案都利用泛型通配符?统一不同Supplier类型，解决编译时类型不兼容问题
• 基于状态码的强制转换是安全的，因为我们已明确对应状态下的解析类型，不会出现运行时类型转换异常
• 方案2将逻辑封装到Handler内部，代码更整洁，符合单一职责原则

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