block()/blockfirst()/blocklast（）在exchange（）之后调用bodyToMono时出现阻塞错误

Question

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

api调用（exchange（））的响应是ClientResponse。在这里，我可以使用bodyToMono转换为ByteArrayResource，该ByteArrayResource可以流式传输到文件，或者，如果在创建文件时出现错误，那么我也可以使用bodyToMono转换为DTO。但是，我似乎不能这样做，也不能依赖于ClientResponse头的内容。

在运行时，我得到一个IllegalStateException

block（）/blockfirst（）/blocklast（）是阻塞的，线程反应器-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();

这可能吗？

Answer 1

首先，一些有助于您理解解决此用例的代码片段的东西。

1. 不应在返回反应类型的方法中调用阻塞方法；您将阻塞应用程序中为数不多的一个线程，这对应用程序非常不利
2. 无论如何，从Reactor 3.2开始，在反应管道中阻塞会引发错误
3. 按照注释中的建议调用subscribe也不是一个好主意。这或多或少类似于在单独的线程中作为任务启动该作业。完成后，您将得到一个回调(subscribe方法可以被赋予lambdas)，但实际上您正在将当前管道与该任务解耦。在这种情况下，在您有机会读取完整的响应正文并将其写入文件之前，可以关闭客户端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 ，这是done（error）回调的反应性等价物，它在事情完成和是否发生错误时发出信号。

因为我不确定createErrorFile方法应该做什么，所以我提供了createsPreadSheet的示例，它只是将正文字节写入文件。请注意，由于Databuffer可能被回收/池化，我们需要在完成后释放它们。

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实例（由于性能原因，反应运算符是预取值），并将以反应方式写入字节。