Netty之粘包分包

粘包現象

---

客戶端在一個for循環內連續發送1000個hello給Netty服務器端,  

        Socket socket = new Socket("127.0.0.1", 10101);
        for(int i = 0; i < 1000; i++){
            socket.getOutputStream().write(“hello”.getBytes());
        }        
        socket.close();

 而在服務器端接受到的信息並不是預期的1000個獨立的Hello字符串.

實際上是無序的hello字符串混合在一起, 如圖所示. 這種現象我們稱之為粘包.

為什么會出現這種現象呢? TCP是個”流”協議，流其實就是沒有界限的一串數據。 

TCP底層中並不了解上層業務數據的具體含義，它會根據TCP緩沖區的實際情況進行包划分，

所以在TCP中就有可能一個完整地包會被TCP拆分成多個包，也有可能吧多個小的包封裝成一個大的數據包發送。

分包處理

---

顧名思義, 我們要對傳輸的數據進行分包. 一個簡單的處理邏輯是在發送數據包之前, 先用四個字節占位, 表示數據包的長度. 

數據包結構為:

|    長度(4字節)    |    數據    |

 

        Socket socket = new Socket("127.0.0.1", 10101);
        String message = "hello";
        byte[] bytes = message.getBytes();
        ByteBuffer buffer = ByteBuffer.allocate(4 + bytes.length);
        // 消息長度
        buffer.putInt(bytes.length);
        // 消息正文
        buffer.put(bytes);
        byte[] array = buffer.array();
        for(int i = 0; i < 1000; i++){
            socket.getOutputStream().write(array);
        }
        socket.close();

 

服務器端代碼, 我們需要借助於FrameDecoder類來分包.

public class MyDecoder extends FrameDecoder {

    @Override
    protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {

        if(buffer.readableBytes() > 4){            
            //標記
            buffer.markReaderIndex();
            //長度
            int length = buffer.readInt();
            
            if(buffer.readableBytes() < length){
                buffer.resetReaderIndex();
                //緩存當前剩余的buffer數據，等待剩下數據包到來
                return null;
            }
            
            //讀數據
            byte[] bytes = new byte[length];
            buffer.readBytes(bytes);
            //往下傳遞對象
            return new String(bytes);
        }
        //緩存當前剩余的buffer數據，等待剩下數據包到來
        return null;
    }

}

 

如此一來, 我們再次在服務器端接受到的消息就是按序打印的hello了.

這邊可能有個疑問, 為什么MyDecoder中數據沒有讀取完畢, 需要return null,

正常的pipeline在數據處理完都是要sendUpstream, 給下一個pipeline的.

這個需要看下FrameDecoder.messageReceived 的源碼. 他在其中緩存了一個cumulation對象, 

如果return了null, 他會繼續往緩存里寫數據來實現分包

    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        Object m = e.getMessage();
        if (!(m instanceof ChannelBuffer)) {
        　　// 數據讀完了, 轉下一個pipeline
            ctx.sendUpstream(e);
        } else {
            ChannelBuffer input = (ChannelBuffer)m;
            if (input.readable()) {
                if (this.cumulation == null) {
                    try {
                        this.callDecode(ctx, e.getChannel(), input, e.getRemoteAddress());
                    } finally {
                        this.updateCumulation(ctx, input);
                    }
                } else {
            　　　　 // 緩存上一次沒讀完整的數據
                    input = this.appendToCumulation(input);

                    try {
                        this.callDecode(ctx, e.getChannel(), input, e.getRemoteAddress());
                    } finally {
                        this.updateCumulation(ctx, input);
                    }
                }

            }
        }
    }

 

那么是不是這樣就萬事大吉了呢?

Socket字節流攻擊

---

在上述代碼中, 我們會在服務器端為客戶端發送的數據包長度, 預先分配byte數組.

如果遇到惡意攻擊, 傳入的數據長度與內容 不匹配. 例如聲明數據長度為Integer.MAX_VALUE.

這樣會消耗大量的服務器資源生成byte[], 顯然是不合理的.

因此我們還要加個最大長度限制.

          if(buffer.readableBytes() > 2048){
             buffer.skipBytes(buffer.readableBytes());
          }

新的麻煩也隨之而來, 雖然可以跳過指定長度, 但是數據包本身就亂掉了.

因為長度和內容不匹配, 跳過一個長度后, 不知道下一段數據的開頭在哪里了.

因此我們自定義數據包里面, 不僅要引入數據包長度, 還要引入一個包頭來划分各個包的范圍.

包頭用任意一段特殊字符標記即可, 例如$$$.

        // 防止socket字節流攻擊
        if(buffer.readableBytes() > 2048){
        　　buffer.skipBytes(buffer.readableBytes());
        }
        // 記錄包頭開始的index
        int beginReader = buffer.readerIndex();
        
        while(true) {
            if(buffer.readInt() == ConstantValue.FLAG) {
                break;
            }
        }

 

新的數據包結構為:

|    包頭(4字節)    |    長度(4字節)    |    數據    |

Netty自帶拆包類

---

自己實現拆包雖然可以細粒度控制, 但是也會有些不方便, 可以直接調用Netty提供的一些內置拆包類.

• FixedLengthFrameDecoder 按照特定長度組包
• DelimiterBasedFrameDecoder 按照指定分隔符組包, 例如本文中的$$$
• LineBasedFrameDecoder 按照換行符進行組包, \r \n等等
• ......