粘包、拆包表現形式

現在假設客戶端向服務端連續傳送了兩個資料包，用packet1和packet2來表示，那麼服務端收到的資料可以分為三種，現列舉如下：

第一種情況，接收端正常收到兩個資料包，即沒有發生拆包和粘包的現象，此種情況不在本文的討論範圍內。

第二種情況，接收端只收到一個數據包，由於TCP是不會出現丟包的，所以這一個資料包中包含了傳送端傳送的兩個資料包的資訊，這種現象即為粘包。這種情況由於接收端不知道這兩個資料包的界限，所以對於接收端來說很難處理。

第三種情況，這種情況有兩種表現形式，如下圖。接收端收到了兩個資料包，但是這兩個資料包要麼是不完整的，要麼就是多出來一塊，這種情況即發生了拆包和粘包。這兩種情況如果不加特殊處理，對於接收端同樣是不好處理的。

粘包問題的解決策略

由於底層的TCP無法理解上層的業務資料，所以在底層是無法保證資料包不被拆分和重組的，這個問題只能通過上層的應用協議棧設計來解決。業界的主流協議的解決方案，可以歸納如下： 
1. 訊息定長，報文大小固定長度，例如每個報文的長度固定為200位元組，如果不夠空位補空格； 
2. 包尾新增特殊分隔符，例如每條報文結束都添加回車換行符（例如FTP協議）或者指定特殊字元作為報文分隔符，接收方通過特殊分隔符切分報文區分； 
3. 將訊息分為訊息頭和訊息體，訊息頭中包含表示資訊的總長度（或者訊息體長度）的欄位； 
4. 更復雜的自定義應用層協議。

Netty粘包和拆包解決方案

Netty提供了多個解碼器，可以進行分包的操作，分別是： 
* LineBasedFrameDecoder （換行）
   LineBasedFrameDecoder是回車換行解碼器，如果使用者傳送的訊息以回車換行符作為訊息結束的標識，則可以直接使用Netty的LineBasedFrameDecoder對訊息進行解碼，只需要在初始化Netty服務端或者客戶端時將LineBasedFrameDecoder正確的新增到ChannelPipeline中即可，不需要自己重新實現一套換行解碼器。
   LineBasedFrameDecoder的工作原理是它依次遍歷ByteBuf中的可讀位元組，判斷看是否有“\n”或者“\r\n”，如果有，就以此位置為結束位置，從可讀索引到結束位置區間的位元組就組成了一行。它是以換行符為結束標誌的解碼器，支援攜帶結束符或者不攜帶結束符兩種解碼方式，同時支援配置單行的最大長度。如果連續讀取到最大長度後仍然沒有發現換行符，就會丟擲異常，同時忽略掉之前讀到的異常碼流。防止由於資料報沒有攜帶換行符導致接收到ByteBuf無限制積壓，引起系統記憶體溢位。


* DelimiterBasedFrameDecoder（新增特殊分隔符報文來分包） 
   DelimiterBasedFrameDecoder是分隔符解碼器，使用者可以指定訊息結束的分隔符，它可以自動完成以分隔符作為碼流結束標識的訊息的解碼。
   回車換行解碼器實際上是一種特殊的DelimiterBasedFrameDecoder解碼器。


* FixedLengthFrameDecoder（使用定長的報文來分包） 
    FixedLengthFrameDecoder是固定長度解碼器，它能夠按照指定的長度對訊息進行自動解碼，開發者不需要考慮TCP的粘包/拆包等問題，非常實用。
    對於定長訊息，如果訊息實際長度小於定長，則往往會進行補位操作，它在一定程度上導致了空間和資源的浪費。但是它的優點也是非常明顯的，編解碼比較簡單，因此在實際專案中仍然有一定的應用場景。

* LengthFieldBasedFrameDecoder (自定義解碼器跟編碼器)

   本文介紹的重點LengthFieldBasedFrameDecoder，一般包含了訊息頭（head）、訊息體（body）：訊息頭是固定的長度，一般有有以下資訊 -> 是否壓縮（zip）、訊息型別（type or cmdid）、訊息體長度（body length）；訊息體長度不是固定的，其大小由訊息頭記載，一般記載業務互動資訊。

  netty對應來說就是編碼器（Encoder）跟解碼器（Decoder），一般其中會有一個基本訊息類對外輸出，egg：

/**
 * @describe 訊息快取區
 * @author zhikai.chen
 * @date 2018年4月28日 上午10:13:35
 */
public class Message {
	
	/**
	 * 要傳送的資料
	 */
	private String data;
	
	/**
	 * 業務編號
	 */
	private short cmdId;
	
	/**
	 * 訊息型別  0xAF 表示心跳包    0xBF 表示超時包  0xCF 業務資訊包
	 */
	private byte type;
	
	/**
	 * 是否壓縮，1是，0不是
	 */
	private	byte zip = 0 ;
	
    /**
     * 封裝要傳送的資料包
     * @param data 業務資料
     * @param cmdId 業務標識號
     * @param type 訊息型別
     */
	public Message(String data,short cmdId,byte type){
		this.data=data;
		this.cmdId=cmdId;
		this.type=type;
	}

	public String getData() {
		return data;
	}

	public void setData(String data) {
		this.data = data;
	}

	public short getCmdId() {
		return cmdId;
	}

	public void setCmdId(short cmdId) {
		this.cmdId = cmdId;
	}

	public byte getType() {
		return type;
	}

	public void setType(byte type) {
		this.type = type;
	}

	public byte getZip() {
		return zip;
	}

	public void setZip(byte zip) {
		this.zip = zip;
	}

Encoder：

/**
 * @describe 訊息編碼器,封裝
 * @author zhikai.chen
 * @date 2018年4月28日 上午10:17:52
 */
public class MessageEncoder extends MessageToByteEncoder<Message> {
	
    // 編碼格式
	private final Charset charset = Charset.forName("UTF-8");
	// 需要壓縮的長度
	private final int compressLength=1024;
	
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
    	String source=msg.getData();
		byte[] body=source.getBytes(charset);
		if(body.length > compressLength){
			msg.setZip((byte)1);
			// 加壓
			body=ZipTool.compress(body);
		}
		
		//cmdId(2)+type(1)+zip(1)+body(4)=8
		//out = Unpooled.directBuffer(8+body.length);
		
		//cmdId
		out.writeShort(msg.getCmdId());
		//type
		out.writeByte(msg.getType());
		//是否加壓
		out.writeByte(msg.getZip());
		//長度
		out.writeInt(body.length);
		//內容
		out.writeBytes(body);
    }
    
}

NioSocketServerInitializer（服務端跟客戶端都一致）：

    //TODO 參考Message
    //body(4)+zip(1)+cmdId(2)+type(1)=8
    //最大長度
    private static final int MAX_FRAME_LENGTH = 1024 * 1024;  
    //這個值就是MessageEncoder body.length(4)
    private static final int LENGTH_FIELD_LENGTH = 4; 
    //這個值就是MessageEncoder zip(1)+cmdId(2)+type(1)
    private static final int LENGTH_FIELD_OFFSET = 4;  
    private static final int LENGTH_ADJUSTMENT = 0;  
    private static final int INITIAL_BYTES_TO_STRIP = 0;

        //TODO 粘包處理
        //解碼器
        ch.pipeline().addLast(new MessageDecoder(MAX_FRAME_LENGTH,LENGTH_FIELD_LENGTH,LENGTH_FIELD_OFFSET,LENGTH_ADJUSTMENT,INITIAL_BYTES_TO_STRIP));
        //編碼器
        ch.pipeline().addLast(new MessageEncoder());

稍微解釋一下：
1）LENGTH_FIELD_LENGTH指的就是我們這邊CustomMsg中length這個屬性的大小，我們這邊是int型，所以是4
2）LENGTH_FIELD_OFFSET指的就是我們這邊length欄位的起始位置，因為前面有zip(1)+cmdId(2)+type(1)=4，所以是4
3）LENGTH_ADJUSTMENT指的是length這個屬性的值，假如我們的body長度是40，有時候，有些人喜歡將length寫成44，因為length本身還佔有4個位元組，

    這樣就需要調整一下，那麼就需要-4，我們這邊沒有這樣做，所以寫0就可以了   

Decoder

/**
 * @describe 訊息解碼器
 * @author zhikai.chen
 * @date 2018年4月28日 上午11:09:15
 */
public class MessageDecoder extends LengthFieldBasedFrameDecoder {
	
    //body(4)+zip(1)+cmdId(2)+type(1)=8
    private static final int HEADER_SIZE = 8;
    
    // 編碼格式
 	private final Charset charset = Charset.forName("UTF-8");

    public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
    }
    
    public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength,int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength,lengthAdjustment,initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        if (in == null) {
            return null;
        }

        // 訊息頭讀取不完整，不做解析返回null，直到讀完整為止
        if (in.readableBytes() <= HEADER_SIZE) {
            return null;
        }

        in.markReaderIndex();

        short cmdId = in.readShort();
        byte type = in.readByte();
        byte zip = in.readByte();
        int dataLength = in.readInt();

        // TODO 網路訊號不好，沒有接收到完整資料
        if (in.readableBytes() < dataLength) {
        	//儲存當前讀到的資料，下一次繼續讀取
        	//斷包處理：檢視ByteToMessageDecoder的channelRead方法，ByteBuf cumulation屬性
            in.resetReaderIndex();
            return null;
        }

        byte[] data = new byte[dataLength];
        in.readBytes(data);
        
        // TODO 手動釋放記憶體
        //in.release(); // or ReferenceCountUtil.release(in);
        
        //判斷是否壓縮
        if(zip==1){
        	data=ZipTool.uncompress(data);
        }
        String body = new String(data, charset);
        Message msg = new Message(body, cmdId, type);
        return msg;
    }
}

Decoder的解碼順序是跟Encoder一致的。

對應的訊息接收handler read處理：

NioSocketServerHandler server=new NioSocketServerHandler();
pipeline.addLast(server);

public class NioSocketServerHandler extends SimpleChannelInboundHandler<Message> {
	
    private final Logger log=LoggerFactory.getLogger(this.getClass());

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
    	log.info("server read msg:{}", JSON.toJSONString(msg));
    }}

write處理：

Message msg=new Message("ok",ModuleID.HEART_BEAT,(byte)0xAF);
ctx.channel().writeAndFlush(msg);

斷包處理

眼尖的童靴其實已經發現(Decoder):

//TODO 訊息頭都讀取不完整，不做解析返回null，直到讀完整為之
        if (in.readableBytes() <= HEADER_SIZE) {
            return null;
        }

        in.markReaderIndex();

        short cmdId = in.readShort();
        byte type = in.readByte();
        byte zip = in.readByte();
        int dataLength = in.readInt();

        // TODO 網路訊號不好，沒有接收到完整資料
        if (in.readableBytes() < dataLength) {
        	//儲存當前讀到的資料，下一次繼續讀取
        	//斷包處理：檢視ByteToMessageDecoder的channelRead方法，ByteBuf cumulation屬性
            in.resetReaderIndex();
            return null;
        }

   說明註釋已經講解了，我們來看看訊息體一次讀不完（斷包）了，netty底層是怎麼處理的：解碼器繼承LengthFieldBasedFrameDecoder，而LengthFieldBasedFrameDecoder繼承ByteToMessageDecoder，ByteToMessageDecoder中有一個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) {
                    // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                    // See https://github.com/netty/netty/issues/4275
                    numReads = 0;
                    discardSomeReadBytes();
                }

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

我們來看看cumulation的定義，

    ByteBuf cumulation;
    private Cumulator cumulator = MERGE_CUMULATOR;
    private boolean singleDecode;
    private boolean decodeWasNull;
    private boolean first;
    private int discardAfterReads = 16;
    private int numReads;

cumulation為ByteBuf物件，是一個緩衝區，即如果訊息體一次讀不完，下一次繼續讀取，直到讀完整訊息頭給定的長度為止。