将单声道转换为无阻塞的 Pojo

Question

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

我找到的所有例子都返回Mono<Pojo>但我必须得到Pojo本身。目前，我正在通过在 Pojo 上调用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 */)
...

我强烈建议您研究一下 Project Reactor 文档中的转换现有序列。

Answer 2

由于不建议使用 Webclient.block((，因此从传入的 httpresponse 中检索值的另一种方法是在具有必填字段的调用应用程序中创建一个 POJO。然后，一旦收到 Mono，使用 Mono.subscribe((，在订阅中添加一个 lambda 函数，输入说 x，使用 x.getters(( 检索各个字段。这些值可以打印在控制台上或分配给本地 var 以进行进一步处理。这在两个方面有所帮助：-

1. 避免可怕的 .block((
2. 在拉取大量数据时保持调用异步。 这是实现预期结果的许多其他方法之一。