Micronaut RxStreamingHttpClient流式传输大文件报错排查与优化咨询

错误日志 #

16:13:20.287 [default-nioEventLoopGroup-1-4] WARN  io.netty.util.ReferenceCountUtil - Failed to release a message: PooledSlicedByteBuf(freed)
io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
        at io.netty.util.internal.ReferenceCountUpdater.toLiveRealRefCnt(ReferenceCountUpdater.java:83)
        at io.netty.util.internal.ReferenceCountUpdater.release(ReferenceCountUpdater.java:148)
        at io.netty.buffer.AbstractReferenceCountedByteBuf.release(AbstractReferenceCountedByteBuf.java:101)
        at io.netty.util.ReferenceCountUtil.release(ReferenceCountUtil.java:90)
        at io.netty.util.ReferenceCountUtil.safeRelease(ReferenceCountUtil.java:116)
        at io.netty.channel.ChannelOutboundBuffer.remove(ChannelOutboundBuffer.java:282)

curl传输结果 #

nwer@alpha ~/tmp $ curl --location 'http://localhost:8080/binary/test1' -i -o test
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current Dload  Upload   Total   Spent    Left  Speed
100 12489    0 12489    0     0   8642      0 --:--:--  0:00:01 --:--:--  8678
curl: (18) transfer closed with outstanding read data remaining

控制器代码 #

@Get(value = "/test1", produces = MediaType.APPLICATION_OCTET_STREAM)
Flowable<ByteBuffer<?>> test1() throws MalformedURLException {
    URL url = new URL("http://localhost:8085");
    HttpClientConfiguration conf = new DefaultHttpClientConfiguration();
    conf.setMaxContentLength(2147483647);
    RxStreamingHttpClient client = RxStreamingHttpClient.create(url, conf);
    return client.dataStream(HttpRequest.GET("/v1/objects/9e3aad86-491c-451c-afbc-98f8c21a6ab8/100MbFile"));
}

问题 #

请问我的实现缺少了什么？这是Micronaut流式传输约100MB文件的最优方式吗？

解答 #

问题根源 #

你当前代码的核心问题是每次请求都手动创建RxStreamingHttpClient实例，这种做法会导致两个严重问题：

1. 客户端实例没有被正确关闭，Netty的ByteBuf资源无法正常回收，引发引用计数错误（就是你看到的IllegalReferenceCountException）。
2. 频繁创建销毁客户端会浪费系统资源，同时导致连接提前关闭，触发curl的“传输中断”错误。

修复方案 #

改用Micronaut的依赖注入管理RxStreamingHttpClient，框架会自动处理客户端的生命周期和资源释放：

1. 注入客户端实例：通过@Client注解注入配置好的流式HTTP客户端。
2. 全局配置参数：将maxContentLength等配置放到application.yml中，避免重复创建配置实例。

修正后的控制器代码：

import io.micronaut.http.MediaType;
import io.micronaut.http.annotation.Get;
import io.micronaut.http.client.RxStreamingHttpClient;
import io.micronaut.http.client.annotation.Client;
import io.micronaut.http.HttpRequest;
import io.reactivex.Flowable;
import java.nio.ByteBuffer;

@Controller("/binary")
public class BinaryController {

    private final RxStreamingHttpClient streamingHttpClient;

    // 依赖注入指定服务的流式客户端
    public BinaryController(@Client("http://localhost:8085") RxStreamingHttpClient streamingHttpClient) {
        this.streamingHttpClient = streamingHttpClient;
    }

    @Get(value = "/test1", produces = MediaType.APPLICATION_OCTET_STREAM)
    public Flowable<ByteBuffer> test1() {
        return streamingHttpClient.dataStream(HttpRequest.GET("/v1/objects/9e3aad86-491c-451c-afbc-98f8c21a6ab8/100MbFile"));
    }
}

对应的application.yml配置：

micronaut:
  http:
    client:
      max-content-length: 2147483647 # 2GB，覆盖默认限制

最优流式传输方式 #

对于100MB级别的大文件，你的Reactive流式思路是对的，但可以再做以下优化：

• 保留Reactive类型输出：Flowable<ByteBuffer>是Micronaut推荐的流式传输方式，能实现逐块传输，避免内存溢出。
• 启用分块传输：Micronaut默认支持HTTP分块传输，只要produces设置为APPLICATION_OCTET_STREAM即可自动启用。
• 配置缓冲区大小：可以在客户端配置中调整chunkSize，平衡内存占用和传输效率。
• 避免手动管理客户端：始终依赖框架的依赖注入来管理HTTP客户端，这是Micronaut的最佳实践。

修正后，Netty的引用计数错误和curl的传输中断问题都会解决，同时实现高效的大文件流式传输。

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