Netty 源碼分析之拆包器的奧祕

爲何要粘包拆包

爲何要粘包

首先你得了解一下TCP/IP協議，在用戶數據量很是小的狀況下，極端狀況下，一個字節，該TCP數據包的有效載荷很是低，傳遞100字節的數據，須要100次TCP傳送，100次ACK，在應用及時性要求不高的狀況下，將這100個有效數據拼接成一個數據包，那會縮短到一個TCP數據包，以及一個ack，有效載荷提升了，帶寬也節省了

非極端狀況，有可能兩個數據包拼接成一個數據包，也有可能一個半的數據包拼接成一個數據包，也有可能兩個半的數據包拼接成一個數據包

爲何要拆包

拆包和粘包是相對的，一端粘了包，另一端就須要將粘過的包拆開，舉個栗子，發送端將三個數據包粘成兩個TCP數據包發送到接收端，接收端就須要根據應用協議將兩個數據包從新組裝成三個數據包

還有一種狀況就是用戶數據包超過了mss(最大報文長度)，那麼這個數據包在發送的時候必須拆分紅幾個數據包，接收端收到以後須要將這些數據包粘合起來以後，再拆開

拆包的原理

在沒有netty的狀況下，用戶若是本身須要拆包，基本原理就是不斷從TCP緩衝區中讀取數據，每次讀取完都須要判斷是不是一個完整的數據包

1.若是當前讀取的數據不足以拼接成一個完整的業務數據包，那就保留該數據，繼續從tcp緩衝區中讀取，直到獲得一個完整的數據包 2.若是當前讀到的數據加上已經讀取的數據足夠拼接成一個數據包，那就將已經讀取的數據拼接上本次讀取的數據，夠成一個完整的業務數據包傳遞到業務邏輯，多餘的數據仍然保留，以便和下次讀到的數據嘗試拼接

netty中拆包的基類

netty 中的拆包也是如上這個原理，內部會有一個累加器，每次讀取到數據都會不斷累加，而後嘗試對累加到的數據進行拆包，拆成一個完整的業務數據包，這個基類叫作 ByteToMessageDecoder，下面咱們先詳細分析下這個類

累加器

ByteToMessageDecoder 中定義了兩個累加器

public static final Cumulator MERGE_CUMULATOR = ...;
public static final Cumulator COMPOSITE_CUMULATOR = ...;
複製代碼

默認狀況下，會使用 MERGE_CUMULATOR

private Cumulator cumulator = MERGE_CUMULATOR;
複製代碼

MERGE_CUMULATOR 的原理是每次都將讀取到的數據經過內存拷貝的方式，拼接到一個大的字節容器中，這個字節容器在 ByteToMessageDecoder中叫作 cumulation

ByteBuf cumulation;
複製代碼

下面咱們看一下 MERGE_CUMULATOR 是如何將新讀取到的數據累加到字節容器裏的

public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
        ByteBuf buffer;
        if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
                || cumulation.refCnt() > 1) {
            buffer = expandCumulation(alloc, cumulation, in.readableBytes());
        } else {
            buffer = cumulation;
        }
        buffer.writeBytes(in);
        in.release();
        return buffer;
}
複製代碼

netty 中ByteBuf的抽象，使得累加很是簡單，經過一個簡單的api調用 buffer.writeBytes(in); 便將新數據累加到字節容器中，爲了防止字節容器大小不夠，在累加以前還進行了擴容處理

static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable) {
        ByteBuf oldCumulation = cumulation;
        cumulation = alloc.buffer(oldCumulation.readableBytes() + readable);
        cumulation.writeBytes(oldCumulation);
        oldCumulation.release();
        return cumulation;
}
複製代碼

擴容也是一個內存拷貝操做，新增的大小便是新讀取數據的大小

拆包抽象

累加器原理清楚以後，下面咱們回到主流程，目光集中在 channelRead 方法，channelRead方法是每次從TCP緩衝區讀到數據都會調用的方法，觸發點在AbstractNioByteChannel的read方法中，裏面有個while循環不斷讀取，讀取到一次就觸發一次channelRead

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        CodecOutputList out = CodecOutputList.newInstance();
        try {
            ByteBuf data = (ByteBuf) msg;
            first = cumulation == null;
            if (first) {
                cumulation = data;
            } else {
                cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
            }
            callDecode(ctx, cumulation, out);
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable t) {
            throw new DecoderException(t);
        } finally {
            if (cumulation != null && !cumulation.isReadable()) {
                numReads = 0;
                cumulation.release();
                cumulation = null;
            } else if (++ numReads >= discardAfterReads) {
                numReads = 0;
                discardSomeReadBytes();
            }

            int size = out.size();
            decodeWasNull = !out.insertSinceRecycled();
            fireChannelRead(ctx, out, size);
            out.recycle();
        }
    } else {
        ctx.fireChannelRead(msg);
    }
}
複製代碼

方法體不長不短，能夠分爲如下幾個邏輯步驟

1.累加數據 2.將累加到的數據傳遞給業務進行業務拆包 3.清理字節容器 4.傳遞業務數據包給業務解碼器處理

###1 累加數據

若是當前累加器沒有數據，就直接跳過內存拷貝，直接將字節容器的指針指向新讀取的數據，不然，調用累加器累加數據至字節容器

ByteBuf data = (ByteBuf) msg;
first = cumulation == null;
if (first) {
    cumulation = data;
} else {
    cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
}
複製代碼

2 將累加到的數據傳遞給業務進行拆包

到這一步，字節容器裏的數據已經是目前未拆包部分的全部的數據了

CodecOutputList out = CodecOutputList.newInstance();
callDecode(ctx, cumulation, out);
複製代碼

callDecode 將嘗試將字節容器的數據拆分紅業務數據包塞到業務數據容器out中

protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    while (in.isReadable()) {
        // 記錄一下字節容器中有多少字節待拆
        int oldInputLength = in.readableBytes();
        decode(ctx, in, out);
        if (out.size() == 0) {
            // 拆包器未讀取任何數據
            if (oldInputLength == in.readableBytes()) {
                break;
            } else {
             // 拆包器已讀取部分數據，還須要繼續
                continue;
            }
        }

        if (oldInputLength == in.readableBytes()) {
            throw new DecoderException(
                    StringUtil.simpleClassName(getClass()) +
                    ".decode() did not read anything but decoded a message.");
        }

        if (isSingleDecode()) {
            break;
        }
    }
}
複製代碼

我將原始代碼作了一些精簡，在解碼以前，先記錄一下字節容器中有多少字節待拆，而後調用抽象函數 decode 進行拆包

protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
複製代碼

netty中對各類用戶協議的支持就體如今這個抽象函數中，傳進去的是當前讀取到的未被消費的全部的數據，以及業務協議包容器，全部的拆包器最終都實現了該抽象方法

業務拆包完成以後，若是發現並無拆到一個完整的數據包，這個時候又分兩種狀況

1.一個是拆包器什麼數據也沒讀取，可能數據還不夠業務拆包器處理，直接break等待新的數據 2.拆包器已讀取部分數據，說明解碼器仍然在工做，繼續解碼

業務拆包完成以後，若是發現已經解到了數據包，可是，發現並無讀取任何數據，這個時候就會拋出一個Runtime異常 DecoderException，告訴你，你什麼數據都沒讀取，卻解析出一個業務數據包，這是有問題的

3 清理字節容器

業務拆包完成以後，只是從字節容器中取走了數據，可是這部分空間對於字節容器來講依然保留着，而字節容器每次累加字節數據的時候都是將字節數據追加到尾部，若是不對字節容器作清理，那麼時間一長就會OOM

正常狀況下，其實每次讀取完數據，netty都會在下面這個方法中將字節容器清理，只不過，當發送端發送數據過快，channelReadComplete可能會好久才被調用一次

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    numReads = 0;
    discardSomeReadBytes();
    if (decodeWasNull) {
        decodeWasNull = false;
        if (!ctx.channel().config().isAutoRead()) {
            ctx.read();
        }
    }
    ctx.fireChannelReadComplete();
}
複製代碼

這裏順帶插一句，若是一次數據讀取完畢以後(可能接收端一邊收，發送端一邊發，這裏的讀取完畢指的是接收端在某個時間再也不接受到數據爲止），發現仍然沒有拆到一個完整的用戶數據包，即便該channel的設置爲非自動讀取，也會觸發一次讀取操做 ctx.read()，該操做會從新向selector註冊op_read事件，以便於下一次能讀到數據以後拼接成一個完整的數據包

因此爲了防止發送端發送數據過快，netty會在每次讀取到一次數據，業務拆包以後對字節字節容器作清理，清理部分的代碼以下

if (cumulation != null && !cumulation.isReadable()) {
    numReads = 0;
    cumulation.release();
    cumulation = null;
} else if (++ numReads >= discardAfterReads) {
    numReads = 0;
    discardSomeReadBytes();
}
複製代碼

若是字節容器當前已無數據可讀取，直接銷燬字節容器，而且標註一下當前字節容器一次數據也沒讀取

若是連續16次（discardAfterReads的默認值），字節容器中仍然有未被業務拆包器讀取的數據，那就作一次壓縮，有效數據段總體移到容器首部

discardSomeReadBytes以前，字節累加器中的數據分佈

+--------------+----------+----------+
|   readed     | unreaded | writable | 
+--------------+----------+----------+
複製代碼

discardSomeReadBytes以後，字節容器中的數據分佈

+----------+-------------------------+
| unreaded |      writable           | 
+----------+-------------------------+
複製代碼

這樣字節容器又能夠承載更多的數據了

4 傳遞業務數據包給業務解碼器處理

以上三個步驟完成以後，就能夠將拆成的包丟到業務解碼器處理了，代碼以下

int size = out.size();
decodeWasNull = !out.insertSinceRecycled();
fireChannelRead(ctx, out, size);
out.recycle();
複製代碼

期間用一個成員變量 decodeWasNull 來標識本次讀取數據是否拆到一個業務數據包，而後調用 fireChannelRead 將拆到的業務數據包都傳遞到後續的handler

static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int numElements) {
    for (int i = 0; i < numElements; i ++) {
        ctx.fireChannelRead(msgs.getUnsafe(i));
    }
}
複製代碼

這樣，就能夠把一個個完整的業務數據包傳遞到後續的業務解碼器進行解碼，隨後處理業務邏輯

行拆包器

下面，以一個具體的例子來看看業netty自帶的拆包器是如何來拆包的

這個類叫作 LineBasedFrameDecoder，基於行分隔符的拆包器，TA能夠同時處理 \n以及\r\n兩種類型的行分隔符，核心方法都在繼承的 decode 方法中

protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    Object decoded = decode(ctx, in);
    if (decoded != null) {
        out.add(decoded);
    }
}
複製代碼

netty 中自帶的拆包器都是如上這種模板，其實能夠加一層，把這這層模板抽取出來的，不知道爲何netty沒有這麼作，咱們接着跟進去，代碼比較長，咱們仍是分模塊來剖析

###1 找到換行符位置

final int eol = findEndOfLine(buffer);

private static int findEndOfLine(final ByteBuf buffer) {
    int i = buffer.forEachByte(ByteProcessor.FIND_LF);
    if (i > 0 && buffer.getByte(i - 1) == '\r') {
        i--;
    }
    return i;
}

ByteProcessor FIND_LF = new IndexOfProcessor((byte) '\n');

複製代碼

for循環遍歷，找到第一個 \n 的位置,若是\n前面的字符爲\r，那就返回\r的位置

###2 非discarding模式的處理 接下來，netty會判斷，當前拆包是否屬於丟棄模式，用一個成員變量來標識

private boolean discarding;
複製代碼

第一次拆包不在discarding模式（ 後面的分支會講何爲非discarding模式），因而進入如下環節

####2.1 非discarding模式下找到行分隔符的處理

// 1.計算分隔符和包長度
final ByteBuf frame;
final int length = eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;

// 丟棄異常數據
if (length > maxLength) {
    buffer.readerIndex(eol + delimLength);
    fail(ctx, length);
    return null;
}

// 取包的時候是否包括分隔符
if (stripDelimiter) {
    frame = buffer.readRetainedSlice(length);
    buffer.skipBytes(delimLength);
} else {
    frame = buffer.readRetainedSlice(length + delimLength);
}
return frame;
複製代碼

1.首先，新建一個幀，計算一下當前包的長度和分隔符的長度（由於有兩種分隔符） 2.而後判斷一下須要拆包的長度是否大於該拆包器容許的最大長度(maxLength)，這個參數在構造函數中被傳遞進來，如超出容許的最大長度，就將這段數據拋棄，返回null 3.最後，將一個完整的數據包取出，若是構造本解包器的時候指定 stripDelimiter爲false，即解析出來的包包含分隔符，默認爲不包含分隔符

####2.2 非discarding模式下未找到分隔符的處理 沒有找到對應的行分隔符，說明字節容器沒有足夠的數據拼接成一個完整的業務數據包，進入以下流程處理

final int length = buffer.readableBytes();
if (length > maxLength) {
    discardedBytes = length;
    buffer.readerIndex(buffer.writerIndex());
    discarding = true;
    if (failFast) {
        fail(ctx, "over " + discardedBytes);
    }
}
return null;
複製代碼

首先取得當前字節容器的可讀字節個數，接着，判斷一下是否已經超過可容許的最大長度，若是沒有超過，直接返回null，字節容器中的數據沒有任何改變，不然，就須要進入丟棄模式

使用一個成員變量 discardedBytes 來表示已經丟棄了多少數據，而後將字節容器的讀指針移到寫指針，意味着丟棄這一部分數據，設置成員變量discarding爲true表示當前處於丟棄模式。若是設置了failFast，那麼直接拋出異常，默認狀況下failFast爲false，即安靜得丟棄數據

###3 discarding模式 若是解包的時候處在discarding模式，也會有兩種狀況發生

####3.1 discarding模式下找到行分隔符

在discarding模式下，若是找到分隔符，那能夠將分隔符以前的都丟棄掉

final int length = discardedBytes + eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
buffer.readerIndex(eol + delimLength);
discardedBytes = 0;
discarding = false;
if (!failFast) {
    fail(ctx, length);
}
複製代碼

計算出分隔符的長度以後，直接把分隔符以前的數據所有丟棄，固然丟棄的字符也包括分隔符，通過這麼一次丟棄，後面就有多是正常的數據包，下一次解包的時候就會進入正常的解包流程

3.2discarding模式下未找到行分隔符

這種狀況比較簡單，由於當前還在丟棄模式，沒有找到行分隔符意味着當前一個完整的數據包還沒丟棄完，當前讀取的數據是丟棄的一部分，因此直接丟棄

discardedBytes += buffer.readableBytes();
buffer.readerIndex(buffer.writerIndex());
複製代碼

特定分隔符拆包

這個類叫作 DelimiterBasedFrameDecoder，能夠傳遞給TA一個分隔符列表，數據包會按照分隔符列表進行拆分，讀者能夠徹底根據行拆包器的思路去分析這個DelimiterBasedFrameDecoder，這裏不在贅述，有問題能夠留言

總結

netty中的拆包過程實際上是和你本身去拆包過程同樣，只不過TA將拆包過程當中邏輯比較獨立的部分抽象出來變成幾個不一樣層次的類，方便各類協議的擴展，咱們平時在寫代碼過程當中，也必須培養這種抽象能力，這樣你的coding水平纔會不斷提升，完。

若是你想系統地學Netty，個人小冊《Netty 入門與實戰：仿寫微信 IM 即時通信系統》能夠幫助你

若是你想系統學習Netty原理，那麼你必定不要錯過個人Netty源碼分析系列視頻：Java 讀源碼之 Netty 深刻剖析