ByteBuf源码

　　ByteBuf是顶层的抽象类，定义了用于传输数据的ByteBuf需要的方法和属性。

AbstractByteBuf

　　直接继承ByteBuf，一些公共属性和方法的公共逻辑会在这里定义。例如虽然不同性质的ByteBuf底层实现不同如堆Buffer和直接Buffer，但在进行数据写入的时候都要检查并移动readIndex，因此诸如检查并移动readIndex的方法就抽取并定义在AbstractByteBuf方法中。

公共成员变量

　　分为两大类，一是读写索引和mark索引，一个是leakDetector，用于检查是否存在内存泄露，该成员变量是static意味着所有ByteBuf的实例共享同一个leakDetector，单例设计模式处处可见。

static final ResourceLeakDetector<ByteBuf> leakDetector;
int readerIndex;
int writerIndex;
private int markedReaderIndex;
private int markedWriterIndex;
private int maxCapacity;

读操作簇

　　以readBytes为例，在AbstractByteBuf里主要做了两件事情

  • 合法性检查
  • 调用getBytes，之所以用调用是因为getBytes会因为子类的不同实现而有所区别
  • 索引移动
    public ByteBuf readBytes(byte[] dst, int dstIndex, int length) {
        this.checkReadableBytes(length);
        this.getBytes(this.readerIndex, dst, dstIndex, length);
        this.readerIndex += length;
        return this;
    }

 　　

写操作簇

　　同样做了三件事情

  • 合法性检查
  • 调用setBytes
  • 索引移动　　
    public ByteBuf writeBytes(ByteBuf src, int srcIndex, int length) {
        this.ensureWritable(length);
        this.setBytes(this.writerIndex, src, srcIndex, length);
        this.writerIndex += length;
        return this;
    }

　　在ensureWritable中封装了扩容逻辑，这也是ByteBuf比NIO中的ByteBuffer优秀的地方。首先检查传入的length是否小于零，小于零则抛出异常，否则执行ensureWritable0，Netty给方法起的名字总是这么随意……

    public ByteBuf ensureWritable(int minWritableBytes) {
        if (minWritableBytes < 0) {
            throw new IllegalArgumentException(String.format("minWritableBytes: %d (expected: >= 0)", minWritableBytes));
        } else {
            this.ensureWritable0(minWritableBytes);
            return this;
        }
    }
  •  如果当前期望写入的数据长度(minWritableByte)大于此时ByteBuf的长度(writableBytes)且大于剩余的最大容量(this.MaxCapacity-this.writerIndex)，抛出异常。ByteBuf真的不能写了。

  •  如果当前期望写入的数据长度(minWritableByte)大于此时ByteBuf的长度(writableBytes)，但小于剩余的最大容量(this.MaxCapacity-this.writerIndex)，调用calculateNewCapacity扩容，返回扩容后的大小(newCapacity)，并重新设置当前ByteBuf(this.capacity(newCapacity))。
  •  如果当前期望写入的数据长度(minWritableByte)小于此时ByteBuf的长度，啥也不干。
final void ensureWritable0(int minWritableBytes) {
        this.ensureAccessible();
        if (minWritableBytes > this.writableBytes()) {
            if (minWritableBytes > this.maxCapacity - this.writerIndex) {
                throw new IndexOutOfBoundsException(String.format("writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s", this.writerIndex, minWritableBytes, this.maxCapacity, this));
            } else {
                int newCapacity = this.alloc().calculateNewCapacity(this.writerIndex + minWritableBytes, this.maxCapacity);
                this.capacity(newCapacity);
            }
        }
    }

　　calculateNewCapacity传入两个参数：能够完成这里写入所需要的最小容量(minNewCapacity)，当前支持的最大容量(maxCapacity)。

  • 首先合法性检查，minNewCapacity<0或者minNewCapacity>maxCapacity抛出异常。
  • 如果需要的容量等于4MB(minNewCapacity==4194304)，那么把4MB作为扩容后的容量并返回。
  • 如果需要的容量大于4MB，首先找到一个小于等于4MB整数倍的容量(newCapacity)，如果newCapacity加上4MB大于maxCapacity则返回maxCapacity，否则返回newCapacity加4MB，保证扩容后的容量永远是4MB的整数倍，这种思想和HashMap扩容很接近，有人把这种扩容称为步进4MB扩容方式。
  • 如果需要的容量小于4MB，从64开始倍增，直到大于等于需要的容量。
public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
        if (minNewCapacity < 0) {
            throw new IllegalArgumentException("minNewCapacity: " + minNewCapacity + " (expected: 0+)");
        } else if (minNewCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format("minNewCapacity: %d (expected: not greater than maxCapacity(%d)", minNewCapacity, maxCapacity));
        } else {
            int threshold = 4194304;
            if (minNewCapacity == 4194304) {
                return 4194304;
            } else {
                int newCapacity;
                if (minNewCapacity > 4194304) {
                    newCapacity = minNewCapacity / 4194304 * 4194304;
                    if (newCapacity > maxCapacity - 4194304) {
                        newCapacity = maxCapacity;
                    } else {
                        newCapacity += 4194304;
                    }

                    return newCapacity;
                } else {
                    for(newCapacity = 64; newCapacity < minNewCapacity; newCapacity <<= 1) {
                    }

                    return Math.min(newCapacity, maxCapacity);
                }
            }
        }
    }

　　这里应该还有一个复制的操作，但是在这个抽象类里没有找到，应该在具体的实现类中。

重用缓冲区

　　把已读的部分[0,readIndex]重用起来。

  • 如果readIndex==0，没有可以重用的区域。
  • 如果readIndex!=0且readIndex!=writeIndex，说明在0~readIndex区间内的内存可以重用，调用setBytes，把尚未读取的内容复制到缓冲区的开始，然后把writerIndex向前挪(this.writerIndex-=this.readerIndex)，重设读索引为0。
  • 如果readIndex!=0且readIndex==writeIndex,说明有可以重用的内存，且不存在还没有读取的数据，直接重设读写索引为0(readIndex=writeIndex=0)
    public ByteBuf discardReadBytes() {
        this.ensureAccessible();
        if (this.readerIndex == 0) {
            return this;
        } else {
            if (this.readerIndex != this.writerIndex) {
                this.setBytes(0, this, this.readerIndex, this.writerIndex - this.readerIndex);
                this.writerIndex -= this.readerIndex;
                this.adjustMarkers(this.readerIndex);
                this.readerIndex = 0;
            } else {
                this.adjustMarkers(this.readerIndex);
                this.writerIndex = this.readerIndex = 0;
            }

            return this;
        }
    }

AbstractReferenceCountedByteBuf

　　直接继承自AbstractByteBuf，添加了引用计数的功能，类似于JVM垃圾回收中的引用计数法，当一个ByteByf对象引用计数为1时，代表该ByteBuf没有有效引用，可以被回收。

属性

  • refCntUpdater，一个Atomic的对象，使用refCntUpdater的CAS方法线程安全的修改refCnt，实现无锁化，提高效率。
  • refCnt，volatile类型的变量，记录当前ByteBuf的引用次数。初始值为1，每次用新的引用增加1，失去一个引用减1，当最终再次回到1的时候释放ByteBuf
private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater =
            AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");

    private volatile int refCnt = 1;


retain

　　每调用一次retain方法refCnt就会线程安全的增加一，retain主要调用了retain0方法，netty方法的名字还是这样随意。熟悉的JUC的味道，熟悉的for(;;)写法，做了两件事情。

  • 合法性检查。nextCnt<=increment就会抛出异常，这里很有意思，分两个情况来看：nextCnt<increment，nextCnt=increment+refCnt，increment是一个大于零的数，只有在溢出的情况下nextCnt才会小于refCnt，所以小于号是来避免溢出的发生。nextCnt=increment，increment一定是大于零的，那么只有refCnt=0的时候才会发生nextCnt=increment，所以在netty看来refCnt=0是一个异常情况，这种异常情况只会发生在错误调用release的情况下才会发生。
  • 自旋CAS修改refCnt。
private ByteBuf retain0(int increment) {
        for (;;) {
            int refCnt = this.refCnt;
            final int nextCnt = refCnt + increment;

            // Ensure we not resurrect (which means the refCnt was 0) and also that we encountered an overflow.
            if (nextCnt <= increment) {
                throw new IllegalReferenceCountException(refCnt, increment);
            }
            if (refCntUpdater.compareAndSet(this, refCnt, nextCnt)) {
                break;
            }
        }
        return this;
    }

release

　　每调用一次release，refCnt会线程安全的减少一，当减少到refCnt==decrement时，调用deallocate方法回收ByteBuf内存，deallocate需要由具体的子类重写。

private boolean release0(int decrement) {
        for (;;) {
            int refCnt = this.refCnt;
            if (refCnt < decrement) {
                throw new IllegalReferenceCountException(refCnt, -decrement);
            }

            if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
                if (refCnt == decrement) {
                    deallocate();
                    return true;
                }
                return false;
            }
        }
    }

UnpooledHeapByteBuf

  • unpooled，非池化，每次使用ByteBuf都需要新申请一个ByteBuf，相较于Pooled效率较低，但使用起来较为简单。
  • Heap，分配在堆上。

属性

  • alloc，聚合了一个ByteBufAllocator，用于分配内存。
  • array，缓冲区。
  • tmpNioBuf，用于实现ByteBuf到NIO中的ByteBuffer转换。
    private final ByteBufAllocator alloc;
    byte[] array;
    private ByteBuffer tmpNioBuf;

capacity

　　用于调整ByteBuf的长度到指定长度，与写方法簇中的动态调整不同，后者的调用对使用者是透明的，而用户可以主动的调用capacity方法调整长度。如果newCapacity>oldCapacity，那么就在ByteBuf里添加一unspecified数据，如果newCapacity<oldCapacity那么从ByteBuf里删除一些数据。

  • 如果newCapacity>oldCapacity，分配一个新的byte数组(allocateArray)，把旧的数组复制过去(arraycopy)，让属性中的arr指向新的数组，释放就的数组对象。
  • 如果newCapacity<oldCapacity，需要截取。如果readIndex<newCapacity<writeCapacity，则把writeIndex设置为新的容量。如果newCapacity<readerIndex，说明没有可读的字节数组需要复制。
    public ByteBuf capacity(int newCapacity) {
        checkNewCapacity(newCapacity);

        int oldCapacity = array.length;
        byte[] oldArray = array;
        if (newCapacity > oldCapacity) {
            byte[] newArray = allocateArray(newCapacity);
            System.arraycopy(oldArray, 0, newArray, 0, oldArray.length);
            setArray(newArray);
            freeArray(oldArray);
        } else if (newCapacity < oldCapacity) {
            byte[] newArray = allocateArray(newCapacity);
            int readerIndex = readerIndex();
            if (readerIndex < newCapacity) {
                int writerIndex = writerIndex();
                if (writerIndex > newCapacity) {
                    writerIndex(writerIndex = newCapacity);
                }
                System.arraycopy(oldArray, readerIndex, newArray, readerIndex, writerIndex - readerIndex);
            } else {
                setIndex(newCapacity, newCapacity);
            }
            setArray(newArray);
            freeArray(oldArray);
        }
        return this;
    }
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/AshOfTime/p/10895178.html