Reactor-Netty HttpClient调用时ByteBuf引用计数异常求助

核心原因 #

Reactor Netty的responseSingle()返回的响应体ByteBuf遵循Netty的引用计数机制，默认会在响应体被“消费”（比如完成map等操作）后自动释放。但在zip（并行）或zipWhen（顺序依赖）操作中，第一个响应的ByteBuf会在第二个请求执行过程中被提前释放——因为第一个响应的流处理已经完成，触发了自动释放逻辑，导致后续访问该ByteBuf时抛出IllegalReferenceCountException。

场景解决方案 #

场景1：依赖型顺序调用（zipWhen） #

方案1：转换为非Netty数据类型（推荐，无需手动管理引用）

将ByteBuf转换为普通字节数组或字符串，脱离Netty的引用计数管理，原ByteBuf会被自动释放，后续处理无需担心资源泄漏：

httpClient.get()
          .uri("/first-api")
          .responseSingle((response, body) -> body.asByteArray().map(bytes -> new Pair<>(response, bytes)))
          .zipWhen(firstPair -> {
              // 从第一个响应头获取参数构造第二个请求
              String param = firstPair.getLeft().header("X-Required-Param");
              return httpClient.get()
                               .uri("/second-api?key=" + param)
                               .responseSingle((resp, body) -> body.asByteArray().map(bytes -> new Pair<>(resp, bytes)));
          })
          .subscribe(result -> {
              Pair<HttpResponse, byte[]> firstResp = result.getT1();
              Pair<HttpResponse, byte[]> secondResp = result.getT2();
              // 处理两个响应的内容
              String firstContent = new String(firstResp.getRight());
              String secondContent = new String(secondResp.getRight());
              // ...
          });

方案2：手动retain+doFinally释放（适合必须保留ByteBuf的场景）

通过retain()增加ByteBuf的引用计数，阻止自动释放，再通过doFinally确保无论处理成功还是失败，都手动释放ByteBuf，避免内存泄漏：

httpClient.get()
          .uri("/first-api")
          .responseSingle((response, body) -> body.retain().map(buf -> new Pair<>(response, buf)))
          .zipWhen(firstPair -> {
              String param = firstPair.getLeft().header("X-Required-Param");
              return httpClient.get()
                               .uri("/second-api?key=" + param)
                               .responseSingle((resp, body) -> body.map(buf -> new Pair<>(resp, buf)));
          })
          .doFinally(signal -> {
              // 手动释放第一个响应的ByteBuf
              result.getT1().getRight().release();
          })
          .subscribe(result -> {
              Pair<HttpResponse, ByteBuf> firstResp = result.getT1();
              Pair<HttpResponse, ByteBuf> secondResp = result.getT2();
              // 处理ByteBuf内容
              // ...
          });

场景2：并行调用合并结果（zip） #

同样适用上述两种方案，优先推荐转换为非Netty类型：

// 定义第一个并行请求
Mono<Pair<HttpResponse, byte[]>> firstCall = httpClient.get()
                                                       .uri("/first-api")
                                                       .responseSingle((resp, body) -> body.asByteArray().map(bytes -> new Pair<>(resp, bytes)));

// 定义第二个并行请求
Mono<Pair<HttpResponse, byte[]>> secondCall = httpClient.get()
                                                        .uri("/second-api")
                                                        .responseSingle((resp, body) -> body.asByteArray().map(bytes -> new Pair<>(resp, bytes)));

// 合并两个请求结果
Mono.zip(firstCall, secondCall)
    .subscribe(result -> {
        Pair<HttpResponse, byte[]> firstResp = result.getT1();
        Pair<HttpResponse, byte[]> secondResp = result.getT2();
        // 合并处理逻辑
        // ...
    });

关键注意事项 #

• 手动retain()后必须配套doFinally释放：无论流处理是成功、失败还是取消，doFinally都会执行，确保ByteBuf被正确释放，不会引发内存泄漏。
• 转换为字节数组/字符串的方式更安全：无需手动管理引用计数，适合数据量不大的场景；如果是大文件等超大数据，建议保留ByteBuf并使用retain+doFinally的方式。

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