Spring Cloud Gateway中RequestDecorator引发内存泄漏的解决方法

问题场景 #

在Spring Cloud Gateway的LoggingFilter中使用自定义RequestDecorator包装请求，代码如下：

public class RequestDecorator extends ServerHttpRequestDecorator {

  private final List<DataBuffer> dataBuffers = new ArrayList<>();

  public RequestDecorator(ServerHttpRequest delegate) {
    super(delegate);
    super.getBody()
        .map(
            dataBuffer -> {
              dataBuffers.add(dataBuffer);
              return dataBuffer;
            })
        .subscribe();
  }

  @Override
  public Flux<DataBuffer> getBody() {
    return copy();
  }

  private Flux<DataBuffer> copy() {
    return Flux.fromIterable(dataBuffers)
        .map(dataBuffer -> dataBuffer.factory().wrap(dataBuffer.asByteBuffer()));
  }
}

性能测试时出现Netty资源泄漏错误：

i.n.u.ResourceLeakDetector               :   - LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
    io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:403)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
    io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
    io.netty.channel.unix.PreferredDirectByteBufAllocator.ioBuffer(PreferredDirectByteBufAllocator.java:53)
    io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:120)
    io.netty.channel.epoll.EpollRecvByteAllocatorHandle.allocate(EpollRecvByteAllocatorHandle.java:75)
    io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:785)
    io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:499)
    io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:397)
    io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
    io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    java.base/java.lang.Thread.run(Thread.java:834)

问题根源 #

原代码直接持有原始DataBuffer实例并缓存到列表中，未正确处理Netty ByteBuf的引用计数：

1. 构造函数中调用subscribe()触发body读取后，原始DataBuffer未被释放，一直被列表持有。
2. copy()方法仅包装原始DataBuffer的ByteBuffer，但原始实例仍未被回收，导致Netty资源泄漏。

修复方案 #

使用Spring提供的DataBufferUtils工具类处理资源释放，核心思路是缓存body的内容副本（而非原始DataBuffer），读取完成后立即释放原始DataBuffer。

方案1：同步缓存（适合小体积请求） #

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import reactor.core.publisher.Flux;
import java.nio.ByteBuffer;

public class RequestDecorator extends ServerHttpRequestDecorator {

    private final ByteBuffer cachedBody;

    public RequestDecorator(ServerHttpRequest delegate) {
        super(delegate);
        // 合并所有DataBuffer并读取内容，自动释放原始资源
        this.cachedBody = DataBufferUtils.join(delegate.getBody())
                .map(dataBuffer -> {
                    ByteBuffer byteBuffer = ByteBuffer.allocate(dataBuffer.readableByteCount());
                    dataBuffer.read(byteBuffer);
                    byteBuffer.flip();
                    // 释放原始DataBuffer
                    DataBufferUtils.release(dataBuffer);
                    return byteBuffer;
                })
                .block();
    }

    @Override
    public Flux<DataBuffer> getBody() {
        if (cachedBody == null) {
            return Flux.empty();
        }
        // 返回缓存内容的副本，避免下游读取影响原始缓存
        return Flux.just(getDelegate().bufferFactory().wrap(cachedBody.duplicate()));
    }
}

方案2：异步缓存（适合大体积请求，符合Reactive模型） #

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import reactor.core.publisher.Flux;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicReference;

public class RequestDecorator extends ServerHttpRequestDecorator {

    private final AtomicReference<ByteBuffer> cachedBodyRef = new AtomicReference<>();

    public RequestDecorator(ServerHttpRequest delegate) {
        super(delegate);
    }

    @Override
    public Flux<DataBuffer> getBody() {
        return Flux.defer(() -> {
            ByteBuffer cachedBody = cachedBodyRef.get();
            if (cachedBody != null) {
                return Flux.just(getDelegate().bufferFactory().wrap(cachedBody.duplicate()));
            }
            // 首次读取时缓存内容并释放原始DataBuffer
            return DataBufferUtils.join(getDelegate().getBody())
                    .flatMapMany(dataBuffer -> {
                        ByteBuffer byteBuffer = ByteBuffer.allocate(dataBuffer.readableByteCount());
                        dataBuffer.read(byteBuffer);
                        byteBuffer.flip();
                        cachedBodyRef.set(byteBuffer);
                        DataBufferUtils.release(dataBuffer);
                        return Flux.just(getDelegate().bufferFactory().wrap(byteBuffer.duplicate()));
                    })
                    .doOnCancel(() -> cachedBodyRef.set(null)); // 取消订阅时清理缓存
        });
    }
}

关键注意事项 #

• 禁止直接持有原始DataBuffer实例，此类对象关联Netty的ByteBuf，必须手动释放。
• 优先使用DataBufferUtils的工具方法处理资源，如release()释放资源、join()合并数据流。
• 缓存body时，需复制底层ByteBuffer或字节数组，避免依赖原始DataBuffer的生命周期。

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