Netty分散式ByteBuf使用的底層實現方式原始碼解析

2022-03-28 16:01:37

概述

熟悉Nio的小夥伴應該對jdk底層byteBuffer不會陌生, 也就是位元組緩衝區, 主要用於對網路底層io進行讀寫, 當channel中有資料時, 將channel中的資料讀取到位元組緩衝區, 當要往對方寫資料的時候, 將位元組緩衝區的資料寫到channel中

但是jdk的byteBuffer是使用起來有諸多不便, 比如只有一個標記位置的指標position, 在進行讀寫操作時要頻繁的通過flip()方法進行指標位置的移動, 極易出錯, 並且byteBuffer的記憶體一旦分配則不能改變, 不支援動態擴容, 當讀寫的內容大於緩衝區記憶體時, 則會發生索引越界異常

而Netty的ByteBuf對jdk的byteBuffer做了重新的定義, 同樣是位元組緩衝區用於讀取網路io中的資料, 但是使用起來大大簡化, 並且支援了自動擴容, 不用擔心讀寫資料大小超過初始分配的大小

byteBuf根據其分類的不同底層實現方式有所不同, 有直接基於jdk底層byteBuffer實現的, 也有基於位元組陣列的實現的, 對於byteBuf的分類, 在後面的小節將會講到

byteBuf中維護了兩個指標, 一是讀指標, 二是寫指標, 兩個指標相互獨立, 在讀操作的時候, 只會移動讀指標, 通過指標位置記錄讀取的位元組數

同樣在寫操作時, 也只會移動寫指標, 通過寫指標的位置記錄寫的位元組數

在每次讀寫操作的過程中都會對指標的位置進行校驗, 讀指標的位置不能超過寫指標, 否則會丟擲異常

同樣, 寫指標不能超過緩衝區分配的記憶體, 則將對緩衝區做擴容操作

具體指標操作, 入下圖所示：

AbstractByteBuf屬性和構造方法

在講AbstractByteBuf之前, 我們首先先了解一下ByteBuf這個類, 這是所有ByteBuf的最頂層抽象, 裡面定義了大量對ByteBuf操作的抽象方法供子類實現

AbstractByteBuf同樣也緩衝區的抽象類, 定義了byteBuf的骨架操作, 比如引數校驗, 自動擴容, 以及一些讀寫操作的指標移動, 但具體的實現, 不同的bytebuf實現起來是不同的, 這種情況則交給其子類實現

AbstractByteBuf繼承了這個類, 並實現了其大部分的方法

首先看這個類的屬性和構造方法

//讀指標
int readerIndex;
//寫指標
int writerIndex;
//儲存讀指標
private int markedReaderIndex;
//儲存寫指標
private int markedWriterIndex;
//最大分配容量
private int maxCapacity;
protected AbstractByteBuf(int maxCapacity) {
    if (maxCapacity < 0) {
        throw new IllegalArgumentException("maxCapacity: " + maxCapacity + " (expected: >= 0)");
    }
    this.maxCapacity = maxCapacity;
}

我們可以看到在屬性中定義了讀寫指標的成員標量, 和讀寫指標位置的儲存

在構造方法中可以傳入可分配的最大記憶體, 然後賦值到成員變數中

我們看幾個最簡單的方法

@Override
public int maxCapacity() {
    return maxCapacity;
}
@Override
public int readerIndex() {
    return readerIndex;
}
@Override
public int writerIndex() {
    return writerIndex;
}

獲取最大記憶體, 獲取讀寫指標這些方法, 對所有的bytebuf都是通用的, 所以可以定義在AbstractByteBuf中

我們以一個writeBytes方法為例, 讓同學們熟悉AbstractByteBuf中哪些部分自己實現, 哪些部分則交給了子類實現:

@Override
public ByteBuf writeBytes(ByteBuf src) {
    writeBytes(src, src.readableBytes());
    return this;
}

這個方法是將源的ByteBuf(引數)中的位元組寫入到自身ByteBuf中

首先這裡呼叫了自身的writeBytes方法, 並傳入引數ByteBuf本身, 以及Bytebuf的可讀位元組數, 我們跟到readbleBytes()方法中, 其實就是呼叫了自身的方法:

@Override
public int readableBytes() {
    return writerIndex - readerIndex;
}

我們看到, 這裡可讀位元組數就是返回了寫指標到讀指標之間的長度

我們再繼續跟到writeBytes(src, src.readableBytes())中:

@Override
public ByteBuf writeBytes(ByteBuf src, int length) {
    if (length > src.readableBytes()) {
        throw new IndexOutOfBoundsException(String.format(
                "length(%d) exceeds src.readableBytes(%d) where src is: %s", length, src.readableBytes(), src));
    }
    writeBytes(src, src.readerIndex(), length);
    src.readerIndex(src.readerIndex() + length);
    return this;
}

這裡同樣呼叫了自身的方法首先會對引數進行驗證, 就是寫入自身的長度不能超過源ByteBuf的可讀位元組數

這裡又呼叫了一個wirte方法, 引數傳入源Bytebuf, 其可讀位元組數, 寫入的長度, 這裡寫入的長度我們知道就是源ByteBuf的可讀位元組數

我們再跟到writeBytes(src, src.readerIndex(), length);

public ByteBuf writeBytes(ByteBuf src, int srcIndex, int length) {
    ensureAccessible();
    ensureWritable(length);
    setBytes(writerIndex, src, srcIndex, length);
    writerIndex += length;
    return this;
}

我們重點關注第二個校驗方法ensureWritable(length)

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

然後我們再跟到ensureWritable0(minWritableBytes)方法中:

private void ensureWritable0(int minWritableBytes) {
    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));
    }
    //自動擴容
    int newCapacity = alloc().calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity);
    capacity(newCapacity);
}

開始做了兩個引數校驗, 第一個表示當前ByteBuf寫入的長度如果要小於可寫位元組數, 則返回

第二個可以換種方式去看minWritableBytes+ writerIndex> maxCapacity 也就是需要寫入的長度+寫指標必須要小於最大分配的記憶體, 否則報錯, 注意這裡最大分配記憶體不帶表當前記憶體, 而是byteBuf所能分配的最大記憶體

如果需要寫入的長度超過了可寫位元組數, 並且需要寫入的長度+寫指標不超過最大記憶體, 則就開始了ByteBuf非常經典也非常重要的操作, 也就是自動擴容

int newCapacity = alloc().calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity);

其中alloc()返回的是當前bytebuf返回的緩衝區分配器物件, 我們之後的小節會講到, 這裡呼叫了其calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity)方法為其擴容, 其中傳入的引數writerIndex + minWritableBytes代表所需要的容量, maxCapacity為最大容量

我們跟到擴容的方法裡面去

public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
    //合法性校驗
    if (minNewCapacity < 0) {
        throw new IllegalArgumentException("minNewCapacity: " + minNewCapacity + " (expectd: 0+)");
    }
    if (minNewCapacity > maxCapacity) {
        throw new IllegalArgumentException(String.format(
                "minNewCapacity: %d (expected: not greater than maxCapacity(%d)", 
                minNewCapacity, maxCapacity));
    }
    //閾值為4mb
    final int threshold = 1048576 * 4;
    //最小需要擴容記憶體(總記憶體) == 閾值
    if (minNewCapacity == threshold) {
        //返回閾值
        return threshold;
    }
    //最小擴容記憶體>閾值
    if (minNewCapacity > threshold) {
        //newCapacity為需要擴容記憶體
        int newCapacity = minNewCapacity / threshold * threshold;
        //目標容量+閾值>最大容量
        if (newCapacity > maxCapacity - threshold) {
            //將最大容量作為新容量
            newCapacity = maxCapacity;
        } else {
            //否則, 目標容量+閾值
            newCapacity += threshold;
        }
        return newCapacity;
    }
    //如果小於閾值
    int newCapacity = 64;
    //目標容量<需要擴容的容量
    while (newCapacity < minNewCapacity) {
        //倍增
        newCapacity <<= 1;
    }
    //目標容量和最大容量返回一個最小的
    return Math.min(newCapacity, maxCapacity);
}

擴容相關的邏輯註釋也寫的非常清楚, 如果小於閾值(4mb), 採用倍增的方式, 如果大於閾值(4mb), 採用平移4mb的方式

我們回到writeBytes(ByteBuf src, int srcIndex, int length):

public ByteBuf writeBytes(ByteBuf src, int srcIndex, int length) {
    ensureAccessible();
    ensureWritable(length);
    setBytes(writerIndex, src, srcIndex, length);
    writerIndex += length;
    return this;
}

再往下看setBytes(writerIndex, src, srcIndex, length), 這裡的引數的意思是從當前byteBuf的writerIndex節點開始寫入, 將源緩衝區src的讀指標位置, 寫lenght個位元組, 這裡的方法中AbstractByteBuf類並沒有提供實現, 因為不同型別的BtyeBuf實現的方式是不一樣的, 所以這裡交給了子類去實現

最後將寫指標後移length個位元組

最後我們回到writeBytes(ByteBuf src, int length)方法中:

public ByteBuf writeBytes(ByteBuf src, int length) {
    if (length > src.readableBytes()) {
        throw new IndexOutOfBoundsException(String.format(
                "length(%d) exceeds src.readableBytes(%d) where src is: %s", length, src.readableBytes(), src));
    }
    writeBytes(src, src.readerIndex(), length);
    src.readerIndex(src.readerIndex() + length);
    return this;
}

當writeBytes(src, src.readerIndex(), length)寫完之後, 通過src.readerIndex(src.readerIndex() + length)將源緩衝區的讀指標後移lenght個位元組

以上對AbstractByteBuf的簡單介紹和其中寫操作的方法的簡單剖析

以上就是Netty分散式ByteBuf使用的底層實現方式原始碼解析的詳細內容，更多關於Netty分散式ByteBuf使用底層實現的資料請關注it145.com其它相關文章！