文章

Netty学习之旅—-ByteBuf源码解读之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf

首先，我们再看一下 ByteBuf 的类设计图，从中更进一步了解ByteBuf。

Netty学习之旅—-ByteBuf源码解读之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf

ByteBuf 继承自 ReferenceCounted,引用计数，也就是说 ByteBuf 的内存回收使用的是引用计数器来实现。

UnpooledHeapByteBuf 是非池化的堆内存实现，而 UnpooledDirectByteBuf 是非池化的堆外内存（直接内存）。非池化的ByteBuf 就是利用完之后就需要销毁，无法重用。

1、UnpooledHeapByteBuf 详解

其继承链：UnpooledHeapByteBuf --> AbstractReferenceCountedByteBuf --> AbstractByteBuf。

1.1 AbstractByteBuf 源码分析

AbstractByteBuf 定义 ByteBuf 的基本属性，诸如 readerIndex, writerIndex, markedReaderIndex, markedWriterIndex, maxCapacity, 我们知道 ByteBuf 的容量是可以自动扩容的。

AbstractByteBuf 的这两个属性，应该引起我们的注意：

  1. SwappedByteBuf swappedBuf
    这个是大端序列与小端序列的转换。
  2. ResourceLeakDetector leakDetector = new ResourceLeakDetector(ByteBuf.class);
    Netty 用来解决内存泄漏检测机制，下一篇会详细介绍。

这里截取一下 SwappedByteBuf 的源码，采用了典型的装饰模式来设计。

  1.     public class SwappedByteBuf extends ByteBuf {
  3.         private final ByteBuf buf;
  5.         private final ByteOrder order;
  7.         public SwappedByteBuf(ByteBuf buf) {
  9.             if (buf == null) {
  11.                 throw new NullPointerException("buf");
  13.             }
  15.             this.buf = buf;
  17.             if (buf.order() == ByteOrder.BIG_ENDIAN) {
  19.                 order = ByteOrder.LITTLE_ENDIAN;
  21.             } else {
  23.                 order = ByteOrder.BIG_ENDIAN;
  25.             }
  27.         }
  29.         @Override
  31.         public ByteOrder order() {
  33.             return order;
  35.         }
  37.         @Override
  39.         public ByteBuf order(ByteOrder endianness) {
  41.             if (endianness == null) {
  43.                 throw new NullPointerException("endianness");
  45.             }
  47.             if (endianness == order) {
  49.                 return this;
  51.             }
  53.             return buf;
  55.         }
  57.     }

关于其他 AbstractByteBuf, 该类设计使用了典型的模板模式，对 ByteBuf 提供的类，实现时提供一种模板，然后再提供一个钩子方法，供子类实现，比如_getLong方法，_setLong等方法，由于该类的实现原理不复杂，就不做进一步的源码解读。

1.2 AbstractReferenceCountedByteBuf

该类主要是实现引用计算的常规方法，充分利用 voliate 内存可见性与 CAS 操作完成 refCnt 变量的维护。

其源码实现如下：

  1.     package io.netty.buffer;
  3.     import io.netty.util.IllegalReferenceCountException;
  4.     import io.netty.util.internal.PlatformDependent;
  6.     import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
  8.     /**
  9.      * Abstract base class for {@link ByteBuf} implementations that count references.
  10.      */
  11.     public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
  13.         private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater;
  15.         static {
  16.             AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> updater =
  17.                     PlatformDependent.newAtomicIntegerFieldUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
  18.             if (updater == null) {
  19.                 updater = AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
  20.             }
  21.             refCntUpdater = updater;
  22.         }
  24.         private volatile int refCnt = 1;
  26.         protected AbstractReferenceCountedByteBuf(int maxCapacity) {
  27.             super(maxCapacity);
  28.         }
  30.         @Override
  31.         public final int refCnt() {
  32.             return refCnt;
  33.         }
  35.         /**
  36.          * An unsafe operation intended for use by a subclass that sets the reference count of the buffer directly
  37.          */
  38.         protected final void setRefCnt(int refCnt) {
  39.             this.refCnt = refCnt;
  40.         }
  42.         @Override
  43.         public ByteBuf retain() {
  44.             for (;;) {
  45.                 int refCnt = this.refCnt;
  46.                 if (refCnt == 0) {
  47.                     throw new IllegalReferenceCountException(0, 1);
  48.                 }
  49.                 if (refCnt == Integer.MAX_VALUE) {
  50.                     throw new IllegalReferenceCountException(Integer.MAX_VALUE, 1);
  51.                 }
  52.                 if (refCntUpdater.compareAndSet(this, refCnt, refCnt + 1)) {
  53.                     break;
  54.                 }
  55.             }
  56.             return this;
  57.         }
  59.         @Override
  60.         public ByteBuf retain(int increment) {
  61.             if (increment <= 0) {
  62.                 throw new IllegalArgumentException("increment: " + increment + " (expected: > 0)");
  63.             }
  65.             for (;;) {
  66.                 int refCnt = this.refCnt;
  67.                 if (refCnt == 0) {
  68.                     throw new IllegalReferenceCountException(0, increment);
  69.                 }
  70.                 if (refCnt > Integer.MAX_VALUE - increment) {
  71.                     throw new IllegalReferenceCountException(refCnt, increment);
  72.                 }
  73.                 if (refCntUpdater.compareAndSet(this, refCnt, refCnt + increment)) {
  74.                     break;
  75.                 }
  76.             }
  77.             return this;
  78.         }
  80.         @Override
  81.         public ByteBuf touch() {
  82.             return this;
  83.         }
  85.         @Override
  86.         public ByteBuf touch(Object hint) {
  87.             return this;
  88.         }
  90.         @Override
  91.         public final boolean release() {
  92.             for (;;) {
  93.                 int refCnt = this.refCnt;
  94.                 if (refCnt == 0) {
  95.                     throw new IllegalReferenceCountException(0, -1);
  96.                 }
  98.                 if (refCntUpdater.compareAndSet(this, refCnt, refCnt - 1)) {
  99.                     if (refCnt == 1) {
  100.                         deallocate();
  101.                         return true;
  102.                     }
  103.                     return false;
  104.                 }
  105.             }
  106.         }
  108.         @Override
  109.         public final boolean release(int decrement) {
  110.             if (decrement <= 0) {
  111.                 throw new IllegalArgumentException("decrement: " + decrement + " (expected: > 0)");
  112.             }
  114.             for (;;) {
  115.                 int refCnt = this.refCnt;
  116.                 if (refCnt < decrement) {
  117.                     throw new IllegalReferenceCountException(refCnt, -decrement);
  118.                 }
  120.                 if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
  121.                     if (refCnt == decrement) {
  122.                         deallocate();
  123.                         return true;
  124.                     }
  125.                     return false;
  126.                 }
  127.             }
  128.         }
  130.         /**
  131.          * Called once {@link #refCnt()} is equals 0.
  132.          */
  133.         protected abstract void deallocate();
  134.     }

该类，我们只需要了解，当一个ByteBuf 被引用的次数为 0 时，dealocate() 方法将被调用，该方法就是具体回收 ByteBuf 的操作，由具体的子类去实现。

1.3 UnpooledHeapByteBuf 与 UnpooledDirectByteBuf

首先该类的内部结构如下：

Netty学习之旅—-ByteBuf源码解读之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf

Netty学习之旅—-ByteBuf源码解读之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf

对于非池化的 UnpooledByteBuf,内部就是使用 array 来存储数据，相对简单，所以源码分析，我还是侧重于UnpooledDirectByteBuf。重点关注如下两个方面：

  1. 容量的扩容
  2. 内存的分配

1.3.1 capacity(int newCapacity)

  1.     public ByteBuf capacity(int newCapacity) {
  2.             ensureAccessible();     // @1 
  3.             if (newCapacity < 0 || newCapacity > maxCapacity()) {
  4.                 throw new IllegalArgumentException("newCapacity: " + newCapacity);
  5.             }
  7.             int readerIndex = readerIndex();
  8.             int writerIndex = writerIndex();
  10.             int oldCapacity = capacity;
  11.             if (newCapacity > oldCapacity) {   // @2
  12.                 ByteBuffer oldBuffer = buffer;
  13.                 ByteBuffer newBuffer = allocateDirect(newCapacity);  //@21
  14.                 oldBuffer.position(0).limit(oldBuffer.capacity());          //@22
  15.                 newBuffer.position(0).limit(oldBuffer.capacity());        //@23
  16.                 newBuffer.put(oldBuffer);                                            //@24
  17.                 newBuffer.clear();                                                         //@25
  18.                 setByteBuffer(newBuffer);                                            //@26
  19.             } else if (newCapacity < oldCapacity) { //@3
  20.                 ByteBuffer oldBuffer = buffer;
  21.                 ByteBuffer newBuffer = allocateDirect(newCapacity);
  22.                 if (readerIndex < newCapacity) {
  23.                     if (writerIndex > newCapacity) {
  24.                         writerIndex(writerIndex = newCapacity);
  25.                     }
  26.                     oldBuffer.position(readerIndex).limit(writerIndex);
  27.                     newBuffer.position(readerIndex).limit(writerIndex);
  28.                     newBuffer.put(oldBuffer);
  29.                     newBuffer.clear();
  30.                 } else {
  31.                     setIndex(newCapacity, newCapacity);
  32.                 }
  33.                 setByteBuffer(newBuffer);
  34.             }
  35.             return this;
  36.         }

代码@1，检测一下访问性，可达性，就是引用数必须大于0，否则该 ByteBuf 的内部空间已经被回收了（堆外内存）。

代码@2，扩容操作，思路新建一个缓存区，然后将原先缓存区的数据全部写入到新的缓存区，然后释放旧的缓存区。

代码@21、22，申请一个直接缓存区，然后将原缓冲区的 postion 设置为0,将 limit 设置为 capacity, 处于释放状态（从缓存区读）。

代码@23，将新缓存区的 postion,limit 属性设置为0，老缓存区 limit。

代码@24，将原缓冲区写入到新的缓存区，然后将缓存区置的 position 设置为0，limt 设置为 capacity,其实这里设置position,capacity 的意义不大，因为 ByteBuf 并不会利用内部的 ByteBuffe r的 limit,postion 属性，而是使用readerIndex, wriateIndex。

代码@26，关联新的 ByteBuffer, 并释放原缓存区的空间。

代码@3，压缩缓存区。实现思路是新建一个缓存区，如果 readerIndex 大于新建的 ByteBuffer 的 capacity,则无需将旧的缓存区内容写入到新的缓存区中。如果 readerIndex 小于新 capacity,那需要将 readerIndex 至( Math.min(writerIndex, newCapacity) )直接的内容写入到新的缓存，然后释放旧的缓存区。

我们在重点关注一下 setByteBuffer(newBuffer) 方法，该方法还负责销毁原先的 ByteBuffer。

  1.     private void setByteBuffer(ByteBuffer buffer) {
  2.             ByteBuffer oldBuffer = this.buffer;
  3.             if (oldBuffer != null) {
  4.                 if (doNotFree) {
  5.                     doNotFree = false;
  6.                 } else {
  7.                     freeDirect(oldBuffer);
  8.                 }
  9.             }
  11.             this.buffer = buffer;
  12.             tmpNioBuf = null;
  13.             capacity = buffer.remaining();
  14.         }

释放原先的内存。

1.3.2 内存分配

Netty 在为内存的分配，单独封装，相关类图：

Netty学习之旅—-ByteBuf源码解读之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf

目前，先关注 UnpooledByteBufAllocator，对象池的ByteBuf在后续章节中重点关注。

结合原代码，有如下两个方法引起了我的注意：

  1. 容量扩容规则（容量增长规则）calculateNewCapacity 方法。
  2. 直接内存的分配。newDirectBuffer 方法。

1.3.2.1 calculateNewCapacity

  1.     public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
  2.             if (minNewCapacity < 0) {
  3.                 throw new IllegalArgumentException("minNewCapacity: " + minNewCapacity + " (expectd: 0+)");
  4.             }
  5.             if (minNewCapacity > maxCapacity) {
  6.                 throw new IllegalArgumentException(String.format(
  7.                         "minNewCapacity: %d (expected: not greater than maxCapacity(%d)",
  8.                         minNewCapacity, maxCapacity));
  9.             }
  10.             final int threshold = 1048576 * 4; // 4 MiB page
  12.             if (minNewCapacity == threshold) {
  13.                 return threshold;
  14.             }
  16.             // If over threshold, do not double but just increase by threshold.
  17.             if (minNewCapacity > threshold) {               //@1
  18.                 int newCapacity = minNewCapacity / threshold * threshold;
  19.                 if (newCapacity > maxCapacity - threshold) {
  20.                     newCapacity = maxCapacity;
  21.                 } else {
  22.                     newCapacity += threshold;
  23.                 }
  24.                 return newCapacity;
  25.             }
  27.             // Not over threshold. Double up to 4 MiB, starting from 64.
  28.             int newCapacity = 64;c   // @2
  29.             while (newCapacity < minNewCapacity) {
  30.                 newCapacity <<= 1;
  31.             }
  33.             return Math.min(newCapacity, maxCapacity);
  34.         }
  • minNewCapacity：本次需要申请的最小内存。
  • macCapacity：最大总内存申请值。

代码@1，如果最小需要的内存超过设置的 threshold（阔值的话），则循环，每次增加threshold,然后看是否达到本次申请目标。

代码@2，如果需要申请的内存小于阔值，则以64个字节以2的幂增长。这里体现了内存扩容时的一个优化点。

1.3.2.2 newDirectBuffer 方法

  1.     @Override
  2.         protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
  3.             ByteBuf buf;
  4.             if (PlatformDependent.hasUnsafe()) {
  5.                 buf = new UnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
  6.             } else {
  7.                 buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
  8.             }
  10.             return toLeakAwareBuffer(buf);
  11.         }

该方法中，除了见到申请一个直接内存外，还将该 buf 变成一个可感知的对象。toLeakAwareBuffer 方法，用于该对象被引用的情况，因为 UnpooledDirectByteBuf 是一个聚合对象，内部维护了一个 java.nio.ByteBuffer 的直接对外内存空间，在什么是释放UnpooledDirectByteBuf 中的堆外内存呢？在 UnpooledDirectByteBuf 被java垃圾回收的时候，应该于此同时需要释放指向的堆外内存，但堆外内存不受JVM GC的管理，所以我们只有感知到UnpooledDirectByteBuf被JVM虚拟机回收后，手动去释放堆外内存，大家想想都知道，我们可以通过JAVA提供的引用机制，来实现跟踪垃圾回收器的收集工作，虚引用的作用来了，下一篇，我将会以这个为入口点，重点分析 Netty 堆外内存如何管理，也就是内存泄露检测等方面的课题。

来源：https://blog.csdn.net/prestigeding/article/details/53977445

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。
netty
netty

# Netty学习之旅—-源码分析Netty内存泄漏检测

2021-5-12 17:15:50

javanetty

ByteBuf 篇之 ByteBuf 内部结构与 API 学习

2021-5-12 17:16:40

0 条回复 A文章作者 M管理员
欢迎您，新朋友，感谢参与互动！
    暂无讨论，说说你的看法吧

请输入验证码

请输入图片中的验证码
点击发送按钮获取验证码

取消
×
×
￥undefined
请打开手机使用微信扫码支付
「」
×

....支付确认中....

「」
×
支付金额

￥

undefined
×
积分支付
您当前的积分为0

检测到您未绑定微信账户，请先绑定微信

立刻绑定
确定
×

打开微信扫一扫

扫码并「关注我们的公众号」安全快捷登录

×

下载海报：