粘包問題的解決策略

由于底層的 TCP 無法理解上層業務數據,所以在底層是無法保證數據包不被拆分和重組的?, 這個問題只能通過上層的應用協議棧設計來解決,根據業界主流的協議的解決方案, 可以歸納如下:

消息定長, 例如每個報文的大小固定長度200字節,如果不夠,空位補齊空格;

在包尾部添加回車換行符進行分割, 例如 FTP 協議;

將消息分為消息頭和消息體,消息頭中包含表示消息總長度(或者消息具體長度)的字段,通常設計思路為消息頭的第一個字段使用 int32 來表示消息的總長度;

更復雜的應用協議層;

?一. LineBasedFrameDecoder 與?StringDecoder

?LineBasedFrameDecoder 與?StringDecoder?的工作原理 ?

LineBasedFrameDecoder?的工作原理是它依次遍歷 ByteBuf 中得可讀字節,判斷看是否有 '\n' 或者 ?'\r\n' , ?如果有,就以此位置為結束位置,從可讀索引到結束位置區間的字節組成一行.他是以換行符為結束標志的解碼器.支持攜帶結束符或者不攜帶結束符兩種編碼方式,同時支持配置單行最大長度 . 如果連續讀取到最大長度后仍然沒有發現換行符,則拋出異常,同時忽略掉之前讀到的異常碼流. StringDecoder 的功能非常簡單,就是將收到的對象轉換成字符串,然后繼續調用后面的handler? LineBasedFrameDecoder + StringDecoder ?組合就是按換行切換的文本解碼器, 它被設計來用于支持 TCP 的粘包和拆包. 使用如下: package time.server.impl;import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.LineBasedFrameDecoder; import io.netty.handler.codec.string.StringDecoder;/*** TODO* * @description* @author mjorcen* @time 2015年5月25日 下午2:50:57*/ public class NTimeServerImpl {public void bind(int port) {// 創建兩個NioEventLoopGroup 實例,NioEventLoopGroup// 是一個線程組,它包含一組NIO線程,專門用于處理網絡事件的處理,實際上他們就是Reactor 線程組// 這里創建兩個的原因是一個用于服務端接收用戶的鏈接,另一個用于進行SocketChannel的網絡讀寫EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {// 創建一個 ServerBootstrap ,它是netty用于NIO服務端的輔助啟動類,目的是降低服務端的開發復雜度.ServerBootstrap bootstrap = new ServerBootstrap();// 設定 服務端接收用戶請求的線程組和用于進行SocketChannel網絡讀寫的線程組 bootstrap.group(bossGroup, workerGroup);// 設置創建的 channel 類型bootstrap.channel(NioServerSocketChannel.class);// 配置 NioServerSocketChannel 的 tcp 參數, BACKLOG 的大小bootstrap.option(ChannelOption.SO_BACKLOG, 1024);// 綁定io處理類（childChannelHandler）.他的作用類似于 reactor 模式中的 handler// 類,主要用于處理網絡 I/O 事件,例如對記錄日志,對消息進行解碼等.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {protected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LineBasedFrameDecoder(1024)); // 加入行處理器ch.pipeline().addLast(new StringDecoder()); // 加入字符串解碼器ch.pipeline().addLast(new TimeServerHandler());}});// 綁定端口,隨后調用它的同步阻塞方法 sync 等等綁定操作成功,完成之后 Netty 會返回一個 ChannelFuture// 它的功能類似于的 Future,主要用于異步操作的通知回調.ChannelFuture channelFuture = bootstrap.bind(port).sync();// 等待服務端監聽端口關閉,調用 sync 方法進行阻塞,等待服務端鏈路關閉之后 main 函數才退出. channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {// 優雅的退出,釋放線程池資源 bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {NTimeServerImpl server = new NTimeServerImpl();server.bind(9091);}}

ServerHandler

package time.server.impl;import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext;import java.util.Date;import time.TimeConfig;/*** TODO* * @description* @author ez* @time 2015年5月25日 下午3:06:09*/ public class TimeServerHandler extends ChannelHandlerAdapter implementsTimeConfig {/** (non-Javadoc)* * @see io.netty.channel.ChannelHandlerAdapter#channelRead(io.netty.channel.* ChannelHandlerContext, java.lang.Object)*/@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {String body = (String) msg;System.out.println("The time server receive order : " + body);String currentTime = QUERY.equalsIgnoreCase(body) ? new Date().toString() : "BAD ORDER";currentTime += System.getProperty("line.separator");System.out.println("currentTime : " + currentTime);ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes("utf-8"));ctx.writeAndFlush(resp);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {cause.printStackTrace();// 當出現異常時,釋放資源. ctx.close();}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}}

?

Client

package time.client.impl;import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.LineBasedFrameDecoder; import io.netty.handler.codec.string.StringDecoder;/*** TODO* * @description* @author ez* @time 2015年5月25日 下午3:17:29*/ public class NTimeClient {public void connect(int port, String host) throws Exception {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group);bootstrap.channel(NioSocketChannel.class);bootstrap.option(ChannelOption.TCP_NODELAY, true);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LineBasedFrameDecoder(1024));ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new TimeClientHandler());}});// 發起異步鏈接操作ChannelFuture future = bootstrap.connect(host, port).sync();// 等待客戶端鏈路關閉 future.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {group.shutdownGracefully();}}public static void main(String[] args) throws Exception {NTimeClient client = new NTimeClient();client.connect(9091, "localhost");} }

ClientHandler

package time.server.impl;import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext;import java.util.Date;import time.TimeConfig;/*** TODO* * @description* @author ez* @time 2015年5月25日 下午3:06:09*/ public class TimeServerHandler extends ChannelHandlerAdapter implementsTimeConfig {/** (non-Javadoc)* * @see io.netty.channel.ChannelHandlerAdapter#channelRead(io.netty.channel.* ChannelHandlerContext, java.lang.Object)*/@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {String body = (String) msg;System.out.println("The time server receive order : " + body);String currentTime = QUERY.equalsIgnoreCase(body) ? new Date().toString() : "BAD ORDER";currentTime += System.getProperty("line.separator");System.out.println("currentTime : " + currentTime);ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes("utf-8"));ctx.writeAndFlush(resp);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {cause.printStackTrace();// 當出現異常時,釋放資源. ctx.close();}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}}

?

2:?DelimiterBasedFrameDecoder
?　　

　　

DelimiterBasedFrameDecoder 跟?LineBasedFrameDecoder 很相似 , 只是增加以自定義的分割符.

　　

ByteBuf buf = Unpooled.copiedBuffer("$".getBytes("utf-8"));ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));

?

3:?FixedLengthFrameDecoder 定長的分割器.? ch.pipeline().addLast(new FixedLengthFrameDecoder(1024));

?

? 以上內容出自 : <Netty ?權威指南>?

轉載于:https://www.cnblogs.com/mjorcen/p/4539205.html

總結

以上是生活随笔為你收集整理的netty 粘包的解决策略的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。