遠端通訊——Buffer

目標：介紹Buffer的相關實現邏輯、介紹dubbo-remoting-api中的buffer包內的原始碼解析。

前言

快取區在NIO框架中非常重要，它作為位元組容器，每個NIO框架都有自己的相應的設計實現。比如Java NIO有ByteBuffer的設計，Mina有IoBuffer的設計，Netty4有ByteBuf的設計。 那麼在本文講到的內容是dubbo對於緩衝區做的一些介面定義，並且做了不同的框架實現緩衝區公共的邏輯。下面是本文要講到的類圖：

接下來我就按照類圖上的一個一個分析，忽略test類。

原始碼分析

（一）ChannelBuffer

該介面繼承了Comparable介面，該介面是通道快取介面，是位元組容器，在netty中也有通道快取的設計，也就是io.netty.buffer.ByteBuf，該介面的方法定義和設計跟ByteBuf幾乎一樣，連註釋都一樣，所以我就不再細說了。

（二）AbstractChannelBuffer

該類實現了ChannelBuffer介面，是通道快取的抽象類，它實現了ChannelBuffer所有方法，但是它實現的方法都是需要被重寫的方法，具體的實現都是需要子類來實現。現在我們來惡補一下這個通道快取的原理，當然這個原理跟netty的ByteBuf原理是分不開的。AbstractChannelBuffer維護了兩個索引，一個用於讀取，另一個用於寫入當你從通道快取中讀取時，readerIndex將會被遞增已經被讀取的位元組數，同樣的當你寫入的時候writerIndex也會被遞增。

/**
 * 讀索引
 */
private int readerIndex;

/**
 * 寫索引
 */
 

private int writerIndex;

/**
 * 標記讀索引
 */
private int markedReaderIndex;

/**
 * 標記寫索引
 */
private int markedWriterIndex;
複製程式碼

可以看到該類有四個屬性，讀索引和寫索引的作用就是我上述介紹的，讀索引和寫索引的起始位置都為索引位置0。而標記讀索引和標記寫索引是為了做備份回滾，當對緩衝區進行讀寫操作時，可能需要對之前的操作進行回滾，我們就需要將當前的讀寫索引備份到相應的標記索引中。

該類的其他方法都是利用四個屬性來操作，無非就是檢測是否有資料可讀或者還是否有空間可寫等方法，做一些前置條件的校驗以及索引的設定，具體的實現都是需要子類來實現，所以我就不貼程式碼，因為邏輯比較簡單。

（三）DynamicChannelBuffer

該類繼承了AbstractChannelBuffer類，該類是動態的通道快取區類，也就是該類是從ChannelBufferFactory工廠中動態的生成緩衝區，預設使用的工廠是HeapChannelBufferFactory。

1.屬性和構造方法

/**
 * 通道快取區工廠
 */
private final ChannelBufferFactory factory;

/**
 * 通道快取區
 */
private ChannelBuffer buffer;

public DynamicChannelBuffer(int estimatedLength) {
    // 預設是HeapChannelBufferFactory
    this(estimatedLength,HeapChannelBufferFactory.getInstance());
}

public DynamicChannelBuffer(int estimatedLength,ChannelBufferFactory factory) {
    // 如果預計長度小於0 則丟擲異常
    if (estimatedLength < 0) {
        throw new IllegalArgumentException("estimatedLength: " + estimatedLength);
    }
    // 如果工廠為空，則丟擲空指標異常
    if (factory == null) {
        throw new NullPointerException("factory");
    }
    // 設定工廠
    this.factory = factory;
    // 建立快取區
    buffer = factory.getBuffer(estimatedLength);
}
複製程式碼

可以看到，該類有兩個屬性，所有的實現方法都是呼叫了buffer的方法，不過該buffer產生是通過工廠動態生成的。並且從構造方法來看，預設使用HeapChannelBufferFactory。

2.ensureWritableBytes

@Override
public void ensureWritableBytes(int minWritableBytes) {
    // 如果最小寫入的位元組數不大於可寫的位元組數，則結束
    if (minWritableBytes <= writableBytes()) {
        return;
    }

    // 新增容量
    int newCapacity;
    // 此緩衝區可包含的位元組數等於0。
    if (capacity() == 0) {
        // 新增容量設定為1
        newCapacity = 1;
    } else {
        // 新增容量設定為緩衝區可包含的位元組數
        newCapacity = capacity();
    }
    // 最小新增容量 = 當前的寫索引+最小寫入的位元組數
    int minNewCapacity = writerIndex() + minWritableBytes;
    // 當新增容量比最小新增容量小
    while (newCapacity < minNewCapacity) {
        // 新增容量左移1位，也就是加倍
        newCapacity <<= 1;
    }

    // 通過工廠建立該容量大小當緩衝區
    ChannelBuffer newBuffer = factory().getBuffer(newCapacity);
    // 從buffer中讀取資料到newBuffer中
    newBuffer.writeBytes(buffer,0,writerIndex());
    // 替換原來到緩衝區
    buffer = newBuffer;
}
複製程式碼

該方法是確保陣列有可寫的容量，該方法是重寫了父類的方法，通過傳入一個最小寫入的位元組數，來對緩衝區進行擴容，可以看到，當現有的緩衝區不夠大的時候，會對緩衝區進行加倍對擴容，直到buffer的大小大於傳入的最小可寫位元組數。

3.copy

@Override
public ChannelBuffer copy(int index,int length) {
    // 建立緩衝區，預計長度最小為64，或者更大
    DynamicChannelBuffer copiedBuffer = new DynamicChannelBuffer(Math.max(length,64),factory());
    // 複製資料
    copiedBuffer.buffer = buffer.copy(index,length);
    // 設定索引，讀索引設定為0，寫索引設定為copy的資料長度
    copiedBuffer.setIndex(0,length);
    // 返回快取區
    return copiedBuffer;
}
複製程式碼

該方法是複製資料，在建立緩衝區的時候，預計長度最小是64，，然後重新設定讀索引寫索引。

其他方法都呼叫了buffer的方法或者呼叫了父類的方法，所以不再這裡多說。

（四）ByteBufferBackedChannelBuffer

該方法繼承AbstractChannelBuffer，該類是基於 Java NIO中的ByteBuffer來實現相關的讀寫資料等操作。

/**
 * ByteBuffer例項
 */
private final ByteBuffer buffer;

/**
 * 容量
 */
private final int capacity;

public ByteBufferBackedChannelBuffer(ByteBuffer buffer) {
    if (buffer == null) {
        throw new NullPointerException("buffer");
    }

    // 建立一個新的位元組緩衝區，新緩衝區的大小將是此緩衝區的剩餘容量
    this.buffer = buffer.slice();
    // 返回buffer的剩餘容量
    capacity = buffer.remaining();
    // 設定寫索引
    writerIndex(capacity);
}
複製程式碼

上述就是該類的屬性和建構函式，可以看到它有一個ByteBuffer型別的例項，並且capacity是buffer的剩餘容量。

還有其他的方法比如getByte方法是從buffer中讀取資料方法，setBytes方法是把資料寫入buffer，它們都有很多過載方法，為就不一一講解了，它們都是呼叫了ByteBuffer中的一些方法，如果對於Java NIO中的ByteBuffer方法不是很熟悉的朋友，需要先了解一下Java NIO中的ByteBuffer。

（五）HeapChannelBuffer

該方法繼承了AbstractChannelBuffer，該類中buffer是基於位元組陣列實現

/**
 * The underlying heap byte array that this buffer is wrapping.
 * 此緩衝區包裝的基礎堆位元組陣列。
 */
protected final byte[] array;

/**
 * Creates a new heap buffer with a newly allocated byte array.
 * 使用新分配的位元組陣列建立新的堆緩衝區。
 *
 * @param length the length of the new byte array
 */
public HeapChannelBuffer(int length) {
    this(new byte[length],0);
}

/**
 * Creates a new heap buffer with an existing byte array.
 * 使用現有位元組陣列建立新的堆緩衝區。
 *
 * @param array the byte array to wrap
 */
public HeapChannelBuffer(byte[] array) {
    this(array,array.length);
}

/**
 * Creates a new heap buffer with an existing byte array.
 * 使用現有位元組陣列建立新的堆緩衝區。
 *
 * @param array       the byte array to wrap
 * @param readerIndex the initial reader index of this buffer
 * @param writerIndex the initial writer index of this buffer
 */
protected HeapChannelBuffer(byte[] array,int readerIndex,int writerIndex) {
    if (array == null) {
        throw new NullPointerException("array");
    }
    this.array = array;
    setIndex(readerIndex,writerIndex);
}
複製程式碼

該類有好幾個建構函式，都是基於位元組陣列的，也就是在該類中包裝了一個位元組陣列，把建構函式傳入的位元組陣列傳入到該屬性中。其他方法邏輯比較簡單。

（六）ChannelBufferFactory

public interface ChannelBufferFactory {

    /**
     * 獲得緩衝區例項
     * @param capacity
     * @return
     */
    ChannelBuffer getBuffer(int capacity);

    ChannelBuffer getBuffer(byte[] array,int offset,int length);

    ChannelBuffer getBuffer(ByteBuffer nioBuffer);

}
複製程式碼

該介面是通道緩衝區工廠，其中就只定義了獲得通道緩衝區的方法，比較好理解，它有兩個實現類，我後續會講到。

（七）HeapChannelBufferFactory

該類實現了ChannelBufferFactory，該類就是基於位元組陣列來建立緩衝區的工廠。

public class HeapChannelBufferFactory implements ChannelBufferFactory {

    /**
     * 單例
     */
    private static final HeapChannelBufferFactory INSTANCE = new HeapChannelBufferFactory();

    public HeapChannelBufferFactory() {
        super();
    }

    public static ChannelBufferFactory getInstance() {
        return INSTANCE;
    }

    @Override
    public ChannelBuffer getBuffer(int capacity) {
        // 建立一個capacity容量的緩衝區
        return ChannelBuffers.buffer(capacity);
    }

    @Override
    public ChannelBuffer getBuffer(byte[] array,int length) {
        return ChannelBuffers.wrappedBuffer(array,offset,length);
    }

    @Override
    public ChannelBuffer getBuffer(ByteBuffer nioBuffer) {
        // 判斷該緩衝區是否有位元組陣列支援
        if (nioBuffer.hasArray()) {
            // 使用
            return ChannelBuffers.wrappedBuffer(nioBuffer);
        }

        // 建立一個nioBuffer剩餘容量的緩衝區
        ChannelBuffer buf = getBuffer(nioBuffer.remaining());
        // 記錄下nioBuffer的位置
        int pos = nioBuffer.position();
        // 寫入資料到buf
        buf.writeBytes(nioBuffer);
        // 把nioBuffer的位置重置到pos
        nioBuffer.position(pos);
        return buf;
    }

}
複製程式碼

該類利用了單例模式，其中的方法比較簡單，就是呼叫了ChannelBuffers中的方法，呼叫的方法實際上還是使用了HeapChannelBuffer中建立緩衝區的方法。

（八）DirectChannelBufferFactory

該類實現了ChannelBufferFactory介面，是直接緩衝區工廠，用來建立直接緩衝區。

public class DirectChannelBufferFactory implements ChannelBufferFactory {

    /**
     * 單例
     */
    private static final DirectChannelBufferFactory INSTANCE = new DirectChannelBufferFactory();

    public DirectChannelBufferFactory() {
        super();
    }

    public static ChannelBufferFactory getInstance() {
        return INSTANCE;
    }

    @Override
    public ChannelBuffer getBuffer(int capacity) {
        if (capacity < 0) {
            throw new IllegalArgumentException("capacity: " + capacity);
        }
        if (capacity == 0) {
            return ChannelBuffers.EMPTY_BUFFER;
        }
        // 生成直接緩衝區
        return ChannelBuffers.directBuffer(capacity);
    }

    @Override
    public ChannelBuffer getBuffer(byte[] array,int length) {
        if (array == null) {
            throw new NullPointerException("array");
        }
        if (offset < 0) {
            throw new IndexOutOfBoundsException("offset: " + offset);
        }
        if (length == 0) {
            return ChannelBuffers.EMPTY_BUFFER;
        }
        if (offset + length > array.length) {
            throw new IndexOutOfBoundsException("length: " + length);
        }

        ChannelBuffer buf = getBuffer(length);
        buf.writeBytes(array,length);
        return buf;
    }

    @Override
    public ChannelBuffer getBuffer(ByteBuffer nioBuffer) {
        // 如果nioBuffer不是隻讀，並且它是直接緩衝區
        if (!nioBuffer.isReadOnly() && nioBuffer.isDirect()) {
            // 建立一個緩衝區
            return ChannelBuffers.wrappedBuffer(nioBuffer);
        }

        // 建立一個nioBuffer剩餘容量的緩衝區
        ChannelBuffer buf = getBuffer(nioBuffer.remaining());
        // 記錄下nioBuffer的位置
        int pos = nioBuffer.position();
        // 寫入資料到buf
        buf.writeBytes(nioBuffer);
        // 把nioBuffer的位置重置到pos
        nioBuffer.position(pos);
        return buf;
    }
複製程式碼

該類中的實現方式與HeapChannelBufferFactory中的實現方式差不多，唯一的區別就是它建立的是一個直接緩衝區。

（九）ChannelBuffers

該類是緩衝區的工具類，提供建立、比較 ChannelBuffer 等公用方法。我在這裡舉兩個方法來講：

public static ChannelBuffer wrappedBuffer(ByteBuffer buffer) {
    // 如果緩衝區沒有剩餘容量
    if (!buffer.hasRemaining()) {
        return EMPTY_BUFFER;
    }
    // 如果是位元組陣列生成的緩衝區
    if (buffer.hasArray()) {
        // 使用buffer的位元組陣列生成一個新的緩衝區
        return wrappedBuffer(buffer.array(),buffer.arrayOffset() + buffer.position(),buffer.remaining());
    } else {
        // 基於ByteBuffer建立一個緩衝區（利用buffer的剩餘容量建立）
        return new ByteBufferBackedChannelBuffer(buffer);
    }
}
複製程式碼

該方法通過buffer來建立一個新的緩衝區。可以看出來呼叫的就是上述生成緩衝區的三個類中的方法，ChannelBuffers中很多方法都是這樣去實現的，邏輯比較簡單。

public static boolean equals(ChannelBuffer bufferA,ChannelBuffer bufferB) {
    // 獲得bufferA的可讀資料
    final int aLen = bufferA.readableBytes();
    // 如果兩個緩衝區的可讀資料大小不一樣，則不是同一個
    if (aLen != bufferB.readableBytes()) {
        return false;
    }

    final int byteCount = aLen & 7;

    // 獲得兩個比較的緩衝區的讀索引
    int aIndex = bufferA.readerIndex();
    int bIndex = bufferB.readerIndex();

    // 最多比較緩衝區中的7個資料
    for (int i = byteCount; i > 0; i--) {
        // 一旦有一個資料不一樣，則不是同一個
        if (bufferA.getByte(aIndex) != bufferB.getByte(bIndex)) {
            return false;
        }
        aIndex++;
        bIndex++;
    }

    return true;
}
複製程式碼

該方法就是比較兩個緩衝區是否為同一個，重寫了equals。

（十）ChannelBufferOutputStream

該類繼承了OutputStream

1.屬性和構造方法

/**
 * 緩衝區
 */
private final ChannelBuffer buffer;
/**
 * 記錄開始寫入的索引
 */
private final int startIndex;

public ChannelBufferOutputStream(ChannelBuffer buffer) {
    if (buffer == null) {
        throw new NullPointerException("buffer");
    }
    this.buffer = buffer;
    // 把開始寫入資料的索引記錄下來
    startIndex = buffer.writerIndex();
}
複製程式碼

該類中包裝了一個緩衝區物件和startIndex，startIndex是記錄開始寫入的索引。

2.writtenBytes

public int writtenBytes() {
    return buffer.writerIndex() - startIndex;
}
複製程式碼

該方法是返回寫入了多少資料。

該類裡面還有write方法，都是呼叫了buffer.writeBytes。

（十一）ChannelBufferInputStream

該類繼承了InputStream

1.屬性和建構函式

/**
 * 緩衝區
 */
private final ChannelBuffer buffer;
/**
 * 記錄開始讀資料的索引
 */
private final int startIndex;
/**
 * 結束讀資料的索引
 */
private final int endIndex;

public ChannelBufferInputStream(ChannelBuffer buffer) {
    this(buffer,buffer.readableBytes());
}

public ChannelBufferInputStream(ChannelBuffer buffer,int length) {
    if (buffer == null) {
        throw new NullPointerException("buffer");
    }
    if (length < 0) {
        throw new IllegalArgumentException("length: " + length);
    }
    if (length > buffer.readableBytes()) {
        throw new IndexOutOfBoundsException();
    }

    this.buffer = buffer;
    // 記錄開始讀資料的索引
    startIndex = buffer.readerIndex();
    // 設定結束讀資料的索引
    endIndex = startIndex + length;
    // 標記讀索引
    buffer.markReaderIndex();
}
複製程式碼

該類裡麵包裝了讀開始索引和結束索引，並且在構造方法中初始化這些屬性。

2.readBytes

public int readBytes() {
    return buffer.readerIndex() - startIndex;
}
複製程式碼

該方法是返回讀了多少資料。

3.available

@Override
public int available() throws IOException {
    return endIndex - buffer.readerIndex();
}
複製程式碼

該方法是返回還剩多少資料沒讀

4.read

@Override
public int read() throws IOException {
    if (!buffer.readable()) {
        return -1;
    }
    return buffer.readByte() & 0xff;
}

@Override
public int read(byte[] b,int off,int len) throws IOException {
    // 判斷是否還有資料可讀
    int available = available();
    if (available == 0) {
        return -1;
    }

    // 獲得需要讀取的資料長度
    len = Math.min(available,len);
    buffer.readBytes(b,off,len);
    return len;
}

複製程式碼

該方法是讀資料，返回讀了資料長度。

5.skip

@Override
public long skip(long n) throws IOException {
    if (n > Integer.MAX_VALUE) {
        return skipBytes(Integer.MAX_VALUE);
    } else {
        return skipBytes((int) n);
    }
}

private int skipBytes(int n) throws IOException {
    int nBytes = Math.min(available(),n);
    // 跳過一些資料
    buffer.skipBytes(nBytes);
    return nBytes;
}

複製程式碼

該方法是跳過n長度來讀資料。

後記

該部分相關的原始碼解析地址：github.com/CrazyHZM/in…

該文章講解了Buffer的相關實現邏輯,我很多方法都沒有貼出原始碼，因為很多都是基於Java NIO的ByteBuffer都設計實現，並且要注意AbstractChannelBuffer的三個子類，也就是生成緩衝區的三種形式，還有就是要注意兩個建立緩衝區例項的工廠。下一篇我會講解telnet部分。如果我在哪一部分寫的不夠到位或者寫錯了，歡迎給我提意見。