ByteBuf源码分析（三十三）

冷不防 2022-10-29 14:23 231阅读 0赞

今天进行ByteBuf源码分析：

一、前言简介：

当我们进行数据传输的时候，往往需要使用到缓冲区，常用的缓冲区就是  JDK NIO  类库提供的 java.nio.Buffer 。实际上，7  种基础类型 (Boolean  除外）都有自己的缓冲区实现。对于  NIO  编程而言，我们主要使用的是 ByteBuffer 。从功能角度而言， ByteBuffer  完全可以满足  NIO  编程的需要，但是由于 NIO  编程的复杂性， ByteBuffer  也有其局限性，它的主要缺点如下。

( 1) ByteBuffer  长度固定，一旦分配完成，它的容量不能动态扩展和收缩，当需要编码的 POJO  对象大于  ByteBuffer  的容量时，会发生索引越界异常 ;

( 2) ByteBuffer  只有一个标识位置的指针  position ，读写的时候需要手工调用  flip() 和 rewind()等，使用者必须小心谨慎地处理这些  API ，否则很容易导致程序处理失败 ;

(3)ByteBuffer  的  API  功能有限，一些高级和实用的特性它不支持，需要使用者自己编程实现。

为了弥补这些不足， Netty  提供了自己的  ByteBuffer  实现—— ByteBuf

**二、ByteBuf**  **的简介**

1、首先， ByteBuf  依然是个  Byte  数组的缓冲区，它的基本功能应该与  JDK  的  ByteBuffer  一致，提供以下几类基本功能。

7  种  Java  基础类型、byte 数组、ByteBuffer (ByteBuf〉等的读写 ; 缓冲区自身的  copy  和  slice 等; 设置网络字节序; 构造缓冲区实例; 操作位置指针等方法。

2、由于  JDK  的  ByteBuffer  已经提供了这些基础能力的实现，因此， Netty ByteBuf  的实现可以有两种策略。

1）、参考  JDK ByteBuffer  的实现，增加额外的功能，解决原  ByteBuffer  的缺点 ;

2）、聚合  JDK ByteBuffer ，通过门面模式对其进行包装，可以减少自身的代码量，降低实现成本。

Netty  里两种方式都用了。

**三、ByteBuf**  **的类层次关系**

类所在的包目录：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

由于  ByteBuf  的实现非常繁杂，因此我们不会对其所有子类都进行穷举分析，我们挑选ByteBuf 的主要接口实现类和主要方法进行分析说明。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

1、从内存分配的角度看， ByteBuf  可以分为两类。

(1 ）堆内存（ HeapByteBuf ）字节缓冲区 : 特点是内存的分配和回收速度快，可以被  JVM 自动回收; 缺点就是如果进行  Socket  的  I/O  读写，需要额外做一次内存复制，将堆内存对应的缓冲区复制到内核 Channel  中，性能会有一定程度的下降。

(2 ）直接内存 (DirectByteBuf) 字节缓冲区 : 非堆内存，它在堆外进行内存分配，相比于堆内存，它的分配和回收速度会慢一些，但是将它写入或者从 Socket Channel  中读取时，由于少了一次内存复制，速度比堆内存快。

正是因为各有利弊, 所以  Netty  提供了多种  ByteBuf  供开发者使用，经验表明， ByteBuf 的最佳实践是在 IO  通信线程的读写缓冲区使用  DirectByteBuf ，后端业务消息的编解码模块使用 HeapByteBuf ，这样组合可以达到性能最优。

2、从内存回收角度看， ByteBuf  也分为两类：

基于对象池的  ByteBuf  和普通  ByteBuf 。两者的主要区别就是基于对象池的  ByteBuf  可以重用 ByteBuf  对象，它自己维护了一个内存池，可以循环利用创建的  ByteBuf ，提升内存的使用效率，降低由于高负载导致的频繁 GC 。测试表明使用内存池后的  Netty  在高负载、大并发的冲击下内存和 GC  更加平稳。 尽管推荐使用基于内存池的 ByteBuf ，但是内存池的管理和维护更加复杂，使用起来也需要更加谨慎，因此，Netty  提供了灵活的策略供使用者来做选择。

其实， ByteBuf  的分类还有一种，对内存的操作方式：我们知道，jdk 提供了底层的  Unsafe 类进行读写，通过 Unsafe  类可以直接拿到对象的内存地址，并且基于这个内存地址进行读写操作。Netty 也提供了相关的 ByteBuf  的实现， UnpooledUnsafeDirectByteBuf 、 PooledUnsafeDirectByteBuf、 UnpooledUnsafeHeapByteBuf 、 PooledUnsafeHeapByteBuf ，基本上都是直接调用 unsafe  对象的  native API ，根据数组对象与偏移量来获取数据，效率相对而言较高。

通过前面 Netty 的使用中，我们已经知道，在我们的程序中获得 ByteBuf 的实例 Netty提供了两种方式 ByteBufAllocator 接口和 Unpooled 工具类，对它们的了解也是必须的。

**四、** **ByteBufAllocator**  **的类层次关系**

ByteBufAllocator  类的层次关系如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

1、 在具体的实现类 UnpooledByteBufAllocator 内部使用了 ByteBuf 的实例化方法。而 Unpooled 实际上在内部进行 ByteBuf 的实例化时，也是使用的 UnpooledByteBufAllocator 来进行的。

public final class Unpooled {
    
        private static final ByteBufAllocator ALLOC = UnpooledByteBufAllocator.DEFAULT;
    
        /**
         * Big endian byte order.
         */
        public static final ByteOrder BIG_ENDIAN = ByteOrder.BIG_ENDIAN;
    
        /**
         * Little endian byte order.
         */
        public static final ByteOrder LITTLE_ENDIAN = ByteOrder.LITTLE_ENDIAN;
    }

PooledByteBufAllocator 相对来说要特殊一点。

**2、AbstractByteBuf**  **的源码**

ByteBuf  里都是抽象方法，所以我们首先看  AbstractByteBuf ，  ByteBuf  的一些公共属性和功能会在 AbstractByteBuf  中实现，下面我们对其属性和重要代码进行分析解读。

***1)成员变量***

首先，像读索引、写索引、 mark 、最大容量等公共属性需要定义，相关 源码：

static final ResourceLeakDetector<ByteBuf> leakDetector =
            ResourceLeakDetectorFactory.instance().newResourceLeakDetector(ByteBuf.class);
    
    int readerIndex;
    int writerIndex;
    private int markedReaderIndex;
    private int markedWriterIndex;
    private int maxCapacity;

我们发现，在 AbstractByteBuf 中并没有定义 ByteBuf 的缓冲区实现，例如 byte 数组或者 DirectByteBuffer。原因显而易见，因为 AbstractByteBuf 并不清楚子类到底是基于堆内存还是直接内存，因此无法提前定义。

***2)读操作***

无论子类如何实现  ByteBuf ，例如  UnpooledHeapByteBuf  使用  byte  数组表示字节缓冲区，UnpooledDirectByteBuf 直接使用  ByteBuffer ，它们的功能都是相同的，操作的结果是等价的。 因此，读操作以及其他的一些公共功能都由父类实现，差异化功能由子类实现，这也就是抽象和继承的价值所在。当然与读操作相关的方法很多，我们选择性的看看 readBytes(byte\[\] dst, int dstIndex, int length)

@Override
    public ByteBuf readBytes(byte[] dst, int dstIndex, int length) {
        checkReadableBytes(length);
        getBytes(readerIndex, dst, dstIndex, length);
        readerIndex += length;
        return this;
    }

在读之前，首先对缓冲区的可用空间进行校验，校验通过之后，调用 getBytes  方法，从当前的读索引开始，复制  length  个字节到目标 byte 数组中。由于不同的子类复制操作的技术实现细节不同，因此该方法由子类实现：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

如果读取成功，需要对读索引进行递增:readerIndex += length 。其他类型的读取操作与之类似。

***3)写操作***

实例化一个  ByteBuf  对象的时候，是可以设置一个  capacity  和一个  maxCapacity ，当writerIndex 达到  capacity  的时候，再往里面写入内容， ByteBuf  就会进行扩容。与读取操作类似，写操作的公共行为在 AbstractByteBuf  中实现。我们选择与读取配套的 writeBytes(byte\[ \] src, int srcIndex, int length) 进行分析，它的功能是将源字节数组中从 srcIndex  开始，到  srcIndex + length  截止的字节数组写入到当前的ByteBuf 中。

@Override
    public ByteBuf writeBytes(byte[] src, int srcIndex, int length) {
        ensureWritable(length);
        setBytes(writerIndex, src, srcIndex, length);
        writerIndex += length;
        return this;
    }

首先对写入字节数组的长度进行合法性校验以及扩容处理。

@Override
    public ByteBuf ensureWritable(int minWritableBytes) {
        if (minWritableBytes < 0) {
            throw new IllegalArgumentException(String.format(
                    "minWritableBytes: %d (expected: >= 0)", minWritableBytes));
        }
        ensureWritable0(minWritableBytes);
        return this;
    }

这里的 minWritableBytes 代表的是写的长度，校验 length 是否 >= 0, 不满足则报错。

final void ensureWritable0(int minWritableBytes) {
        ensureAccessible();
        if (minWritableBytes <= writableBytes()) {
            return;
        }
    
        if (minWritableBytes > maxCapacity - writerIndex) {
            throw new IndexOutOfBoundsException(String.format(
                    "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
                    writerIndex, minWritableBytes, maxCapacity, this));
        }
    
        // Normalize the current capacity to the power of 2.
        int newCapacity = alloc().calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity);
    
        // Adjust to the new capacity.
        capacity(newCapacity);
    }

首先计算并检查容量大小是否超过了 maxCapacity，如果超过了则抛出一个 IndexOutOfBoundsException 异常。calculateNewCapacity()方法则是实现扩容的核心， 这个方法的整体思路是：有一个控制阈值，系统定义为 4M（这是个经验值）。首先还是进行各种检查，如果容量较小，2 的指数倍递增是较为合理的，因为不会造成很大的内存浪费，而且可以减少反复扩容的次数；如果容量比较大时，比如 10M 的空间，进行 2 的指数倍新增后容量为 20M，所以这里的就造成了空间的浪费，一般都是先进行指数倍递增 （64->128->256->512…..），到达阀值后 4M，按阀值的步进递增。

**五、AbstractReferenceCountedByteBuf**

从类的名字就可以看出该类主要是对引用进行计数，类似于 JVM  内存回收的对象引用计数器，用于跟踪对象的分配和销毁，做自动内存回收。在具体的实现上，使用了Java  并发编程里的  CAS  原子类型。

public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
    
        private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater =
                AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
    
        private volatile int refCnt;
    
        protected AbstractReferenceCountedByteBuf(int maxCapacity) {
            super(maxCapacity);
            refCntUpdater.set(this, 1);
        }
    
        @Override
        public int refCnt() {
            return refCnt;
        }

首先看第一个字段 refCntUpdater，它是 AtomicIntegerFieldUpdater 类型变量，通过原子的方式对成员变量进行更新等操作，以实现线程安全，消除锁。最后定义了一个 volatile 修饰的 refCnt 字段用于跟踪对象的引用次数，使用 volatile 是为了解决多线程并发访问的可见性问题，此处不对 volatile 的用法展开说明。每调用一次 retain 方法，引用计数器就会增加，具体的增加则是使用了原子类型的 getAndAdd，这是个线程安全的方法。每调用一次 release 方法，引用计数器就会减少，同时，如果发现引用计数为 0，则会进行缓存的释放 deallocate，这个方法则是由子类负责实现的。

**六、UnpooledHeapByteBuf**

UnpooledHeapByteBuf  是基于堆内存进行内存分配的字节缓冲区，它没有基于对象池技术实现，这就意味着每次 IO  的读写都会创建一个新的  UnpooledHeapByteBuf ，频繁进行大块内存的分配和回收对性能会造成一定影响，但是相比于堆外内存的申请和释放，它的成本还是会低一些。相比于 PooledHeapByteBuf， UnpooledHeapByteBuf  的实现原理更加简单，也不容易出现 内存管理方面的问题，因此在满足性能的情况下，推荐使用 UnpooledHeapByteBuf 。首先看下 UnpooledHeapByteBuf  的成员变量定义

public class UnpooledHeapByteBuf extends AbstractReferenceCountedByteBuf {
    
        private final ByteBufAllocator alloc;
        byte[] array;
        private ByteBuffer tmpNioBuf;

首先，它聚合了一个 ByteBufAllocator，用于 UnpooledHeapByteBuf 的内存分配，紧接着定义了一个 byte 数组作为缓冲区，最后定义了一个 ByteBuffer 类型的 tmpNioBuf 变量用于实现 Netty ByteBuf 到 JDK NIO ByteBuffer 的转换。事实上，如果使用 JDK 的 ByteBuffer 替换 byte 数组也是可行的，直接使用 byte 数组的根本原因就是提升性能和更加便捷地进行位操作。JDK 的 ByteBuffer 底层实现也是 byte 数组。知道了这个类由哪些成员变量，那么成员方法基本上都是围绕着这个 byte 数组进行操作，包括在父类 AbstractByteBuf 中没有实现的 setBytes，getBytes 等等。 转换成 JDK ByteBuffer，其实也很简单。因为 ByteBuf 基于 byte 数组实现，NIO 的 ByteBuffer 提供了 wrap 方法，可以将 byte 数组转换成 ByteBuffer 对象。

@Override
    public ByteBuffer nioBuffer(int index, int length) {
        ensureAccessible();
        return ByteBuffer.wrap(array, index, length).slice();
    }

七、UnpooledDirectByteBuf 源码：

public class UnpooledDirectByteBuf extends AbstractReferenceCountedByteBuf {
    
        private final ByteBufAllocator alloc;
    
        private ByteBuffer buffer;
        private ByteBuffer tmpNioBuf;
        private int capacity;
        private boolean doNotFree;

在成员变量上，我们可以看到 UnpooledDirectByteBuf 直接使用的 JDK 中的 ByteBuffer，具体的实现类则是用的

protected ByteBuffer allocateDirect(int initialCapacity) {
        return ByteBuffer.allocateDirect(initialCapacity);
    }

说明它内部缓冲区由 java.nio.DirectByteBuffer 实现。其他成员方法也基本上是围绕着这个 ByteBuffer 来操作的。

到此、ByteBuf源码分析完成，下篇我们分析ByteBufAllocator 的源码，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/0c268b326a0e4400bc9c1a98dc98a103.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/8efc35c725e645d68a2b1569db526b7d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/5c7a30a6184c41fe901a8c02baafe466.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/327f2b0cef1b4dc096b7c74e619400db.png