Netty-自定義協議 · 小機的專屬后端開發筆記本

# Netty自定義協議使用Netty可以幫助我們很快的自定義屬于自己的協議。說明如何自定義協議之前先來看看Netty是如何解決TCP的粘包和半包問題的。   ## 一、粘包和半包 - 粘包：客戶端發送的多次數據被服務端一次性接受，例如客戶端分別發送“abc”、“efd”、“hijk”；可能被服務器接收的時候一次性接收成“abcefdhijk”。 - 半包：客戶端發送的數據可能會被服務端分多次接收，例如客戶端發送“abcefdhijk”，但是服務端卻接收成“abc”、“efd”、“hijk”。     粘包和半包的問題的本質是TCP協議采用流式傳送，其傳送是以字節為單位的，消息沒有邊界，TCP協議每次發送的大小和接收的大小都會送**滑動窗口**的限制；同時對于應用層讀取傳輸層的數據，并不是每次傳輸層有數據了就一次性讀取，而是有個緩沖區在起作用。     除此之外，粘包現象可能由Nagle算法造成，在發送端中，如果每次發送的字節數目太小，則會根據Nagle算法將字節緩存在緩沖區中，等待后面一次性發送。而半包現象可能是要發送的數據包多大，超過了數據鏈路層的MTU的限制，導致數據鏈路層不得不將發送的數據分成多個數據幀發送出去。   **Nagle 算法** 使用tcp協議，即使發送一個字節，也需要加入 tcp 頭和 ip 頭，也就是總字節數會使用 41 bytes，非常不經濟。因此為了提高網絡利用率，tcp 希望盡可能發送足夠大的數據，這就是 Nagle 算法產生的緣由。該算法是指發送端即使還有應該發送的數據，但如果這部分數據很少的話，則進行延遲發送，具體情況如下： - 如果 SO\_SNDBUF 的數據達到 MSS（MTU - 頭部），則需要發送。 - 如果 SO\_SNDBUF 中含有 FIN（表示需要連接關閉）這時將剩余數據發送，再關閉。 - 如果 TCP\_NODELAY = true，則需要發送。 - 已發送的數據都收到 ack 時，則需要發送。 - 上述條件不滿足，但發生超時（一般為 200ms）則需要發送。 - 除上述情況，延遲發送。   ### 解決粘包和半包問題的方案 1. 短鏈接：即發送一個數據包則創建一條tcp鏈接，這種方式效率很低。   2. 固定長度：每一條消息采用固定的長度，接收端根據協商好的固定長度一次性讀取。 Netty中提供*FixedLengthFrameDecoder*類就是采用固定長度編解碼器來解決粘包和半包的問題。使用： ~~~ ch.pipeline.addLast(new FixedLengthFrameDecoder(8)); # 固定長度為8字節 ~~~ 缺點： - 數據包的長度大小不好把握，長度固定太長則會造成浪費，長度固定太小則會對某些數據包不夠使用。   3. 固定分割符：對每一條消息最后添加分割符、例如‘\n’，‘\r\n’。 Netty中提供了*LineBasedFrameDecoder*來使用‘\n’，或者‘\r\n’來作為分割符。如果要使用自定義的字符的話可以使用*DelimiterBasedFrameDecoder*。使用： ~~~ ch.pipeline().addLast(new LineBasedFrameDecoder(1024)); ~~~ 服務端加入，默認以 \\n 或 \\r\\n 作為分隔符，如果超出指定長度仍未出現分隔符，則拋出異常。缺點：處理字符數據比較合適，但如果內容本身包含了分隔符（字節數據常常會有此情況），那么就會解析錯誤。   4. 預設長度：每一條消息分為 head 和 body，head 中包含 body 的長度，**常用！** Netty提供*LengthFieldBasedFrameDecoder*來支持預設長度的方式解決粘包和半包問題。構造函數： ~~~ public LengthFieldBasedFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {...} ~~~ - maxFrameLength：數據包的最大長度。 - lengthFieldOffset：長度字段的偏移值，即要從接收的數據包的哪個字節開始才算長度字段，一般都設置為0。 - lengthFieldLength：長度字段占多少個字節。 - lengthAdjustment：長度字段為基準，還有幾個字節是內容。 - initialBytesToStrip：從頭開始剝離多少個字節是內容。例如： ~~~ // 表示最大長度為1024個字節，長度字段為第一個節點，并且占1個字節，內容從第一個字節后開始。 ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 1, 0, 1)); ~~~ 客戶端： ~~~ public class HelloWorldClient { static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class); public static void main(String[] args) { NioEventLoopGroup worker = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.channel(NioSocketChannel.class); bootstrap.group(worker); bootstrap.handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { log.debug("connetted..."); ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); ch.pipeline().addLast(new ChannelInboundHandlerAdapter() { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.debug("sending..."); Random r = new Random(); char c = 'a'; ByteBuf buffer = ctx.alloc().buffer(); for (int i = 0; i < 10; i++) { byte length = (byte) (r.nextInt(16) + 1); // 先寫入長度頭部信息 buffer.writeByte(length); // 再寫入數據 body信息 for (int j = 1; j <= length; j++) { buffer.writeByte((byte) c); } c++; } // 所以發送的數據為 length+content，這個length要和服務端的協商好 ctx.writeAndFlush(buffer); } }); } }); ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync(); channelFuture.channel().closeFuture().sync(); } catch (InterruptedException e) { log.error("client error", e); } finally { worker.shutdownGracefully(); } } } ~~~   ## 二、協議的設計與解析其實所謂粘包和半包的問題的本質是如何基于TCP協議設計一個不定長的應用層協議。因為TCP中使基于流的方式傳輸的，消息是沒有邊界的。**所以協議的目的就是劃定消息的邊界，制定通信雙方要遵守的通信規則。** 只要是按照協議的，雙方就能夠通信，例如redis中set方法的協議格式如下： ~~~ set name zhangsan # 其協議的格式如下： *3 \n $3 \n set \n $4 \n name \n $8 \n zhangsan \n ~~~ 則在Netty中可以進行模擬： ~~~ private void set(ChannelHandlerContext ctx) { ByteBuf buf = ctx.alloc().buffer(); buf.writeBytes("*3".getBytes()); buf.writeBytes(LINE); buf.writeBytes("$3".getBytes()); buf.writeBytes(LINE); buf.writeBytes("set".getBytes()); buf.writeBytes(LINE); buf.writeBytes("$3".getBytes()); buf.writeBytes(LINE); buf.writeBytes("aaa".getBytes()); buf.writeBytes(LINE); buf.writeBytes("$3".getBytes()); buf.writeBytes(LINE); buf.writeBytes("bbb".getBytes()); buf.writeBytes(LINE); ctx.writeAndFlush(buf); } ~~~ 連接到對應的redis即可進行通信。   ### 2.1 自定義協議要素 **頭部head**： 1. 魔數：一般在數據包的前面幾個字節，用來第一時間判斷是否是無效數據包。 2. 版本號：可以支持協議的升級。 3. 序列化算法：消息正文到底采用哪種序列化反序列化方式，可以由此擴展，例如：json、protobuf、hessian、jdk。 4. 指令類型：根據具體業務區分指令類型。 5. 請求序號：為了雙工通信雙方提供異步通信的能力。 6. 正文長度 **正文body** 消息正文，json、xml、對象流等。