java - 无阻塞地将 Mono 转换为 Pojo

Question

有没有办法将Mono对象转换为java Pojo？ 我有一个 Web 客户端连接到第 3 方 REST 服务，而不是返回 Mono我必须提取该物体并询问它。

我找到的所有示例都返回 Mono<Pojo>但我必须得到 Pojo 本身。目前，我正在通过调用 block() 来完成此操作。在 Pojo 上，但是有没有更好的方法来避免阻塞？

该 block 的问题是，在几次运行后，它开始抛出一些错误，例如 block 因错误而终止。

 public MyPojo getPojo(){
     return myWebClient.get()
                .uri(generateUrl())
                .headers(createHttpHeaders(headersMap))
                .exchange()
                .flatMap(evaluateResponseStatus())
                .block();
}


private Function<ClientResponse, Mono<? extends MyPojo>> evaluateResponseStatus() {
      return response -> {
            if (response.statusCode() == HttpStatus.OK) {
                return response.bodyToMono(MyPojo.class);
            }
            if (webClientUtils.isError(response.statusCode())) {
                throw myHttpException(response);
                // This invokes my exceptionAdvice
                // but after few runs its ignored and 500 error is returned.
            }
            return Mono.empty();
        };
    }

Answer 1

阻止对响应式流中的值进行操作并不是一个好主意。 Project Reactor 为您提供了一系列操作符供您处理流中的对象。

在您的情况下，您可以编写 getPojo() 方法，如下所示:

public Mono<MyPojo> getPojo() {
     return myWebClient.get()
            .uri(generateUrl())
            .headers(createHttpHeaders(headersMap))
            .retrieve()
            .onStatus(status -> webClientUtils.isError(status), 
                      response -> Mono.error(myHttpException(response))
            .bodyToMono(MyPojo.class)
}

请注意，使用 onStatus 方法，我们替换了示例中的整个 evaluateResponseStatus 方法。

您可以像下面这样使用此方法:

// some method
...
getPojo()
    .map(pojo -> /* do something with the pojo instance */)
...

我强烈建议您查看Transforming an existing sequence在 Project Reactor 文档中。