在Spring Reactive中如何从ServerHttpRequest获取InputStream返回值？

我正在为一个基于传统Servlet的第三方SPI编写Spring Reactive实现，其中有个方法要求返回请求的InputStream：

/**
 * Returns the request input stream
 *
 * @param buffered if the input stream should be buffered and support for multiple reads
 * @return the request input stream
 */
 InputStream getInputStream(boolean buffered);

我的实现代码如下：

private InputStream inputStream;

private ServerHttpRequest request;

@Override
public InputStream getInputStream(boolean buffered) {
  if (inputStream != null) {
    return inputStream;
  }

  var inputStreamMono =
      request
          .getBody()
          .collect(
              InputStreamCollector::new,
              (t, dataBuffer) -> t.collectInputStream(dataBuffer.asInputStream()))
          .flatMap(inputStreamCollector -> Mono.just(inputStreamCollector.getInputStream()));

  if (buffered) {
    return inputStream = new BufferedInputStream(Objects.requireNonNull(inputStreamMono.block()));
  }

  return inputStreamMono.block();
}

public class InputStreamCollector {
  private InputStream is;

  public InputStream getInputStream() {
    return this.is;
  }

  public void collectInputStream(InputStream is) {
    if (this.is == null) this.is = is;
    this.is = new SequenceInputStream(this.is, is);
  }
}

运行时抛出以下异常：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-2
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:87) ~[reactor-core-3.6.4.jar:3.6.4]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    *__checkpoint ⇢ org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ LogoutWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ ServerRequestCacheWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ AuthenticationWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ ReactorContextWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ HttpHeaderWriterWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
    *__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
    *__checkpoint ⇢ HTTP GET "/linkerp/manager" [ExceptionHandlingWebHandler]
Original Stack Trace:
        at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:87) ~[reactor-core-3.6.4.jar:3.6.4]
        at reactor.core.publisher.Mono.block(Mono.java:1779) ~[reactor-core-3.6.4.jar:3.6.4]
        at org.umbrella.ReactiveHttpRequest.getInputStream(ReactiveHttpRequest.java:85) ~[classes/:na]

请问我是不是哪里考虑不周？有没有从响应式ServerHttpRequest返回InputStream的正确方式？

问题根源 #

异常已经明确指出：在Reactor的非阻塞线程（parallel-2）中调用block()是被禁止的。Spring Reactive的核心是异步非阻塞模型，block()会强制线程等待结果，直接打破了这个模型的设计原则，因此会被框架拦截抛出异常。

另外你的InputStreamCollector实现也存在问题：SequenceInputStream在读取时会逐个关闭传入的流，但DataBuffer对应的InputStream绑定到缓冲区资源，直接拼接会导致缓冲区泄漏，且无法保证字节聚合的正确性。

正确实现思路 #

要从响应式请求体生成InputStream，核心是先将所有异步数据流的缓冲区聚合为完整的字节数组，再基于这个字节数组创建InputStream——只有把异步流转换为同步的字节数据后，才能安全生成传统的InputStream。

代码实现 #

private InputStream inputStream;
private ServerHttpRequest request;
// 缓存聚合后的字节数组，避免重复处理请求体
private byte[] cachedBody;

@Override
public InputStream getInputStream(boolean buffered) {
    if (inputStream != null) {
        return inputStream;
    }

    try {
        // 聚合所有DataBuffer为字节数组
        cachedBody = request.getBody()
                .map(dataBuffer -> {
                    byte[] bytes = new byte[dataBuffer.readableByteCount()];
                    dataBuffer.read(bytes);
                    DataBufferUtils.release(dataBuffer); // 必须释放缓冲区资源
                    return bytes;
                })
                .reduce(new byte[0], (acc, bytes) -> {
                    // 合并字节数组
                    byte[] result = new byte[acc.length + bytes.length];
                    System.arraycopy(acc, 0, result, 0, acc.length);
                    System.arraycopy(bytes, 0, result, acc.length, bytes.length);
                    return result;
                })
                // 如果当前线程不允许阻塞，需要切换到允许阻塞的线程池执行
                .subscribeOn(Schedulers.boundedElastic())
                .block();
    } catch (IllegalStateException e) {
        throw new RuntimeException("无法在非阻塞线程中同步获取请求体，请确保方法在允许阻塞的上下文调用", e);
    }

    if (cachedBody == null) {
        cachedBody = new byte[0];
    }

    InputStream result = new ByteArrayInputStream(cachedBody);
    if (buffered) {
        result = new BufferedInputStream(result);
    }
    inputStream = result;
    return inputStream;
}

关键注意事项 #

1. 线程上下文限制：如果SPI是在纯Spring Reactive链路（如Spring WebFlux、Spring Cloud Gateway）中调用，绝对不能直接在响应式线程调用block()。此时要么重构SPI接口为响应式（比如返回Mono<InputStream>），要么通过subscribeOn(Schedulers.boundedElastic())将操作转移到允许阻塞的线程池执行。
2. 资源释放：所有DataBuffer必须调用DataBufferUtils.release()手动释放，否则会引发内存泄漏。
3. 重复读取支持：ByteArrayInputStream天然支持重复读取，配合BufferedInputStream可以提升读取性能，完全满足buffered参数的需求。

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