18 Buffer 缓冲区：我们不生产数据，我们只是数据的搬运工 Buffer 是一种字节容器，在 Netty 等 NIO 框架中都有类似的设计，例如，Java NIO 中的ByteBuffer、Netty4 中的 ByteBuf。Dubbo 抽象出了 ChannelBuffer 接口对底层 NIO 框架中的 Buffer 设计进行统一，其子类如下图所示：

Drawing 0.png

ChannelBuffer 继承关系图

下面我们就按照 ChannelBuffer 的继承结构，从顶层的 ChannelBuffer 接口开始，逐个向下介绍，直至最底层的各个实现类。

ChannelBuffer 接口

ChannelBuffer 接口的设计与 Netty4 中 ByteBuf 抽象类的设计基本一致，也有 readerIndex 和 writerIndex 指针的概念，如下所示，它们的核心方法也是如出一辙。

  • getBytes()、setBytes() 方法：从参数指定的位置读、写当前 ChannelBuffer，不会修改 readerIndex 和 writerIndex 指针的位置。
  • readBytes() 、writeBytes() 方法：也是读、写当前 ChannelBuffer，但是 readBytes() 方法会从 readerIndex 指针开始读取数据，并移动 readerIndex 指针；writeBytes() 方法会从 writerIndex 指针位置开始写入数据，并移动 writerIndex 指针。
  • markReaderIndex()、markWriterIndex() 方法：记录当前 readerIndex 指针和 writerIndex 指针的位置，一般会和 resetReaderIndex()、resetWriterIndex() 方法配套使用。resetReaderIndex() 方法会将 readerIndex 指针重置到 markReaderIndex() 方法标记的位置，resetwriterIndex() 方法同理。
  • capacity()、clear()、copy() 等辅助方法用来获取 ChannelBuffer 容量以及实现清理、拷贝数据的功能，这里不再赘述。
  • factory() 方法：该方法返回创建 ChannelBuffer 的工厂对象，ChannelBufferFactory 中定义了多个 getBuffer() 方法重载来创建 ChannelBuffer，如下图所示，这些 ChannelBufferFactory的实现都是单例的。

Drawing 1.png

ChannelBufferFactory 继承关系图

AbstractChannelBuffer 抽象类实现了 ChannelBuffer 接口的大部分方法，其核心是维护了以下四个索引。

  • readerIndex、writerIndex（int 类型）：通过 readBytes() 方法及其重载读取数据时，会后移 readerIndex 索引；通过 writeBytes() 方法及其重载写入数据的时候，会后移 writerIndex 索引。
  • markedReaderIndex、markedWriterIndex（int 类型）：实现记录 readerIndex（writerIndex）以及回滚 readerIndex（writerIndex）的功能，前面我们已经介绍过markReaderIndex() 方法、resetReaderIndex() 方法以及 markWriterIndex() 方法、resetWriterIndex() 方法，你可以对比学习。

AbstractChannelBuffer 中 readBytes() 和 writeBytes() 方法的各个重载最终会通过 getBytes() 方法和 setBytes() 方法实现数据的读写，这些方法在 AbstractChannelBuffer 子类中实现。下面以读写一个 byte 数组为例，进行介绍： public void readBytes(byte[] dst, int dstIndex, int length) { // 检测可读字节数是否足够 checkReadableBytes(length); // 将readerIndex之后的length个字节数读取到dst数组中dstIndex~ // dstIndex+length的位置 getBytes(readerIndex, dst, dstIndex, length); // 将readerIndex后移length个字节 readerIndex += length; } public void writeBytes(byte[] src, int srcIndex, int length) { // 将src数组中srcIndex~srcIndex+length的数据写入当前buffer中 // writerIndex~writerIndex+length的位置 setBytes(writerIndex, src, srcIndex, length); // 将writeIndex后移length个字节 writerIndex += length; }

Buffer 各实现类解析

了解了 ChannelBuffer 接口的核心方法以及 AbstractChannelBuffer 的公共实现之后，我们再来看 ChannelBuffer 的具体实现。

HeapChannelBuffer 是基于字节数组的 ChannelBuffer 实现，我们可以看到其中有一个 array（byte[]数组）字段，它就是 HeapChannelBuffer 存储数据的地方。HeapChannelBuffer 的 setBytes() 以及 getBytes() 方法实现是调用 System.arraycopy() 方法完成数组操作的，具体实现如下： public void setBytes(int index, byte[] src, int srcIndex, int length) { System.arraycopy(src, srcIndex, array, index, length); } public void getBytes(int index, byte[] dst, int dstIndex, int length) { System.arraycopy(array, index, dst, dstIndex, length); }

HeapChannelBuffer 对应的 ChannelBufferFactory 实现是 HeapChannelBufferFactory，其 getBuffer() 方法会通过 ChannelBuffers 这个工具类创建一个指定大小 HeapChannelBuffer 对象，下面简单介绍两个 getBuffer() 方法重载：

@Override public ChannelBuffer getBuffer(int capacity) { // 新建一个HeapChannelBuffer，底层的会新建一个长度为capacity的byte数组 return ChannelBuffers.buffer(capacity); } @Override public ChannelBuffer getBuffer(byte[] array, int offset, int length) { // 新建一个HeapChannelBuffer，并且会拷贝array数组中offset~offset+lenght // 的数据到新HeapChannelBuffer中 return ChannelBuffers.wrappedBuffer(array, offset, length); }

其他 getBuffer() 方法重载这里就不再展示，你若感兴趣的话可以参考源码进行学习。 DynamicChannelBuffer 可以认为是其他 ChannelBuffer 的装饰器，它可以为其他 ChannelBuffer 添加动态扩展容量的功能。DynamicChannelBuffer 中有两个核心字段：

  • buffer（ChannelBuffer 类型），是被修饰的 ChannelBuffer，默认为 HeapChannelBuffer。
  • factory（ChannelBufferFactory 类型），用于创建被修饰的 HeapChannelBuffer 对象的 ChannelBufferFactory 工厂，默认为 HeapChannelBufferFactory。

DynamicChannelBuffer 需要关注的是 ensureWritableBytes() 方法，该方法实现了动态扩容的功能，在每次写入数据之前，都需要调用该方法确定当前可用空间是否足够，调用位置如下图所示：

Drawing 2.png

ensureWritableBytes() 方法如果检测到底层 ChannelBuffer 对象的空间不足，则会创建一个新的 ChannelBuffer（空间扩大为原来的两倍），然后将原来 ChannelBuffer 中的数据拷贝到新 ChannelBuffer 中，最后将 buffer 字段指向新 ChannelBuffer 对象，完成整个扩容操作。ensureWritableBytes() 方法的具体实现如下： public void ensureWritableBytes(int minWritableBytes) { if (minWritableBytes <= writableBytes()) { return; } int newCapacity; if (capacity() == 0) { newCapacity = 1; } else { newCapacity = capacity(); } int minNewCapacity = writerIndex() + minWritableBytes; while (newCapacity < minNewCapacity) { newCapacity «= 1; } ChannelBuffer newBuffer = factory().getBuffer(newCapacity); newBuffer.writeBytes(buffer, 0, writerIndex()); buffer = newBuffer; }

ByteBufferBackedChannelBuffer 是基于 Java NIO 中 ByteBuffer 的 ChannelBuffer 实现，其中的方法基本都是通过组合 ByteBuffer 的 API 实现的。下面以 getBytes() 方法和 setBytes() 方法的一个重载为例，进行分析：

public void getBytes(int index, byte[] dst, int dstIndex, int length) { ByteBuffer data = buffer.duplicate(); try { // 移动ByteBuffer中的指针 data.limit(index + length).position(index); } catch (IllegalArgumentException e) { throw new IndexOutOfBoundsException(); } // 通过ByteBuffer的get()方法实现读取 data.get(dst, dstIndex, length); } public void setBytes(int index, byte[] src, int srcIndex, int length) { ByteBuffer data = buffer.duplicate(); // 移动ByteBuffer中的指针 data.limit(index + length).position(index); // 将数据写入底层的ByteBuffer中 data.put(src, srcIndex, length); }

ByteBufferBackedChannelBuffer 的其他方法实现比较简单，这里就不再展示，你若感兴趣的话可以参考源码进行学习。

NettyBackedChannelBuffer 是基于 Netty 中 ByteBuf 的 ChannelBuffer 实现，Netty 中的 ByteBuf 内部维护了 readerIndex 和 writerIndex 以及 markedReaderIndex、markedWriterIndex 这四个索引，所以 NettyBackedChannelBuffer 没有再继承 AbstractChannelBuffer 抽象类，而是直接实现了 ChannelBuffer 接口。

NettyBackedChannelBuffer 对 ChannelBuffer 接口的实现都是调用底层封装的 Netty ByteBuf 实现的，这里就不再展开介绍，你若感兴趣的话也可以参考相关代码进行学习。

相关 Stream 以及门面类

在 ChannelBuffer 基础上，Dubbo 提供了一套输入输出流，如下图所示：

Drawing 3.png

ChannelBufferInputStream 底层封装了一个 ChannelBuffer，其实现 InputStream 接口的 read/*() 方法全部都是从 ChannelBuffer 中读取数据。ChannelBufferInputStream 中还维护了一个 startIndex 和一个endIndex 索引，作为读取数据的起止位置。ChannelBufferOutputStream 与 ChannelBufferInputStream 类似，会向底层的 ChannelBuffer 写入数据，这里就不再展开，你若感兴趣的话可以参考源码进行分析。

最后要介绍 ChannelBuffers 这个门面类，下图展示了 ChannelBuffers 这个门面类的所有方法：

Drawing 4.png

对这些方法进行分类，可归纳出如下这些方法。

  • dynamicBuffer() 方法：创建 DynamicChannelBuffer 对象，初始化大小由第一个参数指定，默认为 256。
  • buffer() 方法：创建指定大小的 HeapChannelBuffer 对象。
  • wrappedBuffer() 方法：将传入的 byte[] 数字封装成 HeapChannelBuffer 对象。
  • directBuffer() 方法：创建 ByteBufferBackedChannelBuffer 对象，需要注意的是，底层的 ByteBuffer 使用的堆外内存，需要特别关注堆外内存的管理。

  • equals() 方法：用于比较两个 ChannelBuffer 是否相同，其中会逐个比较两个 ChannelBuffer 中的前 7 个可读字节，只有两者完全一致，才算两个 ChannelBuffer 相同。其核心实现如下示例代码： public static boolean equals(ChannelBuffer bufferA, ChannelBuffer bufferB) { final int aLen = bufferA.readableBytes(); if (aLen != bufferB.readableBytes()) { return false; // 比较两个ChannelBuffer的可读字节数 } final int byteCount = aLen & 7; // 只比较前7个字节 int aIndex = bufferA.readerIndex(); int bIndex = bufferB.readerIndex(); for (int i = byteCount; i > 0; i–) { if (bufferA.getByte(aIndex) != bufferB.getByte(bIndex)) { return false; // 前7个字节发现不同，则返回false } aIndex++; bIndex++; } return true; }

  • compare() 方法：用于比较两个 ChannelBuffer 的大小，会逐个比较两个 ChannelBuffer 中的全部可读字节，具体实现与 equals() 方法类似，这里就不再重复讲述。

总结

本课时重点介绍了 dubbo-remoting 模块 buffers 包中的核心实现。我们首先介绍了 ChannelBuffer 接口这一个顶层接口，了解了 ChannelBuffer 提供的核心功能和运作原理；接下来介绍了 ChannelBuffer 的多种实现，其中包括 HeapChannelBuffer、DynamicChannelBuffer、ByteBufferBackedChannelBuffer 等具体实现类，以及 AbstractChannelBuffer 这个抽象类；最后分析了 ChannelBufferFactory 使用到的 ChannelBuffers 工具类以及在 ChannelBuffer 之上封装的 InputStream 和 OutputStream 实现。

关于本课时，你若还有什么疑问或想法，欢迎你留言跟我分享。

参考资料

https://learn.lianglianglee.com/%e4%b8%93%e6%a0%8f/Dubbo%e6%ba%90%e7%a0%81%e8%a7%a3%e8%af%bb%e4%b8%8e%e5%ae%9e%e6%88%98-%e5%ae%8c/18%20%20Buffer%20%e7%bc%93%e5%86%b2%e5%8c%ba%ef%bc%9a%e6%88%91%e4%bb%ac%e4%b8%8d%e7%94%9f%e4%ba%a7%e6%95%b0%e6%8d%ae%ef%bc%8c%e6%88%91%e4%bb%ac%e5%8f%aa%e6%98%af%e6%95%b0%e6%8d%ae%e7%9a%84%e6%90%ac%e8%bf%90%e5%b7%a5.md