block()/blockFirst()/blockLast() 在调用 bodyToMono AFTER exchange() 时出现阻塞错误

新手上路,请多包涵

我正在尝试使用 Webflux 将生成的文件流式传输到另一个位置,但是,如果文件的生成遇到错误,api 将返回成功,但在生成文件而不是文件本身时会使用 DTO 详细说明错误。这是使用非常陈旧且设计不佳的 api,因此请原谅使用 post 和 api 设计。

api 调用 (exchange()) 的响应是 ClientResponse。从这里我可以使用 bodyToMono 转换为 ByteArrayResource,它可以流式传输到文件,或者,如果在创建文件时出错,那么我也可以使用 bodyToMono 转换为 DTO。但是,我似乎不能做或取决于 ClientResponse 标头的内容。

在运行时我得到一个 IllegalStateException 引起的

block()/blockFirst()/blockLast()是阻塞的,thread reactor-http-client-epoll-12不支持

我认为我的问题是我不能在同一个函数链中调用 block() 两次。

我的代码片段是这样的:

 webClient.post()
        .uri(uriBuilder -> uriBuilder.path("/file/")
                                      .queryParams(params).build())
        .exchange()
        .doOnSuccess(cr -> {
                if (MediaType.APPLICATION_JSON_UTF8.equals(cr.headers().contentType().get())) {
                    NoPayloadResponseDto dto = cr.bodyToMono(NoPayloadResponseDto.class).block();
                    createErrorFile(dto);
                }
                else {
                    ByteArrayResource bAr = cr.bodyToMono(ByteArrayResource.class).block();
                    createSpreadsheet(bAr);
                }
            }
        )
        .block();

基本上我想根据标头中定义的 MediaType 以不同方式处理 ClientResponse。

这可能吗?

原文由 DaithiG 发布,翻译遵循 CC BY-SA 4.0 许可协议

阅读 2k
2 个回答

首先,有几件事可以帮助您理解解决此用例的代码片段。

  1. 你永远不应该在返回反应类型的方法中调用阻塞方法;您将阻塞应用程序的几个线程之一,这对应用程序非常不利
  2. 无论如何,从 Reactor 3.2 开始, 在反应管道内阻塞会引发错误
  3. 正如评论中所建议的那样,调用 subscribe 也不是一个好主意。它或多或少类似于在单独的线程中将作业作为任务启动。完成后您会收到回调(可以为 subscribe 方法提供 lambda),但实际上您正在将当前管道与该任务解耦。在这种情况下,在您有机会读取完整的响应正文并将其写入文件之前,可以关闭客户端 HTTP 响应并清理资源
  4. 如果您不想在内存中缓冲整个响应,Spring 提供 DataBuffer (想想可以合并的 ByteBuffer 实例)。
  5. 如果您正在实施的方法本身是阻塞的(例如返回 void ),您可以调用 block,例如在测试用例中。

这是您可以用来执行此操作的代码片段:

 Mono<Void> fileWritten = WebClient.create().post()
        .uri(uriBuilder -> uriBuilder.path("/file/").build())
        .exchange()
        .flatMap(response -> {
            if (MediaType.APPLICATION_JSON_UTF8.equals(response.headers().contentType().get())) {
                Mono<NoPayloadResponseDto> dto = response.bodyToMono(NoPayloadResponseDto.class);
                return createErrorFile(dto);
            }
            else {
                Flux<DataBuffer> body = response.bodyToFlux(DataBuffer.class);
                return createSpreadsheet(body);
            }
        });
// Once you get that Mono, you should give plug it into an existing
// reactive pipeline, or call block on it, depending on the situation

如您所见,我们没有在任何地方阻塞,并且处理 I/O 的方法正在返回 Mono<Void> ,这是反应式等同于 done(error) 回调,它在事情完成时发出信号,如果发生错误。

由于我不确定 createErrorFile 方法应该做什么,我提供了一个示例 createSpreadsheet 只是将正文字节写入文件。请注意,由于数据缓冲区可能会被回收/合并,因此我们需要在完成后释放它们。

 private Mono<Void> createSpreadsheet(Flux<DataBuffer> body) {
    try {
        Path file = //...
        WritableByteChannel channel = Files.newByteChannel(file, StandardOpenOption.WRITE);
        return DataBufferUtils.write(body, channel).map(DataBufferUtils::release).then();
    } catch (IOException exc) {
        return Mono.error(exc);
    }
}

通过此实现,您的应用程序将在给定时间在内存中保存一些 DataBuffer 实例(出于性能原因,反应性运算符会预取值)并将以反应性方式写入字节。

原文由 Brian Clozel 发布,翻译遵循 CC BY-SA 4.0 许可协议

[更新 2021/10/19]

toProcessor() 现已弃用。

考虑使用

myMono.toFuture().get();


正如投票最多的答案中所述,永远不要阻止。就我而言,这是唯一的选择,因为我们在一段命令式代码中使用反应式库。可以通过 将单声道包装在处理器 中来完成阻塞:

 myMono.toProcessor().block()

原文由 adelinor 发布,翻译遵循 CC BY-SA 4.0 许可协议

撰写回答
你尚未登录,登录后可以
  • 和开发者交流问题的细节
  • 关注并接收问题和回答的更新提醒
  • 参与内容的编辑和改进,让解决方法与时俱进
推荐问题