回收累积字节缓冲区发布器

Question

我们正在迁移到Spring WebFlux（带有Reactornetty）。应用程序使用HTTP协议和Spring控制器。目前，我们有一种过渡解决方案，它将入站IO缓冲区累积到合成字节buf中，而不进行复制（然后将其作为输入流进行处理）。Reactornetty为我们提供直接字节缓冲区。因此，为这些缓冲区调用release（）是至关重要的。最初我们有代码：

public static Mono<CompositeByteBuf> collectToComposite(Publisher<DataBuffer> data) {
  return Flux.from(data).reduce(
      EMPTY,
      (CompositeByteBuf acc, DataBuffer buffer) -> {
        ByteBuf byteBuf = toByteBuf(buffer);
        CompositeByteBuf composite = (acc == EMPTY) ? byteBuf.alloc().compositeBuffer(256) : acc;
        composite.addComponent(true, byteBuf);
        return composite;
      }
  ).map(composite -> composite != EMPTY ? composite : createEmptyComposite());
}

在处理得到的复合缓冲液后进行释放。

但当上游发布器发出错误信号时，这种方法会导致泄漏。因此，在下一次尝试中，我们尝试用这样的方式处理错误并释放缓冲区（省略一些角落案例处理）：

public static Mono<CompositeByteBuf> collectToComposite(Publisher<DataBuffer> data) {
  // such code is not suitable for multiple subscribers
  class CompositeHolder {
    CompositeByteBuf composite;

    void addComponent(ByteBuf component) {
      if (composite == null) {
        composite = component.alloc().compositeBuffer(256);
      }
      composite.addComponent(true, component);
    }
  }
  CompositeHolder holder = new CompositeHolder();
  return Flux.from(data)
      .doOnNext(buffer -> holder.addComponent(toByteBuf(buffer)))
      .doOnError(e -> holder.composite.release())
      .then(Mono.fromSupplier(() -> holder.composite));
}

但在那之后，我们意识到有必要在订阅取消时回收缓冲区（当底层连接关闭时发生）。首先想到的是使用doOn取消运算符，但实际上不能保证我们不能为相同的请求调用doOnError和doOn取消回调。因此，简单的解决方案要求我们明确检查缓冲区之前是否已释放。

现在我被卡住了。我不知道如何处理这个案子，避免额外的复杂性。

Answer 1

doFinally是一个doOn操作符，每当源代码出错、完成或取消订阅时都会调用它。此外，它保证回调只执行一次（在错误取消的情况下）。

您提供消费者