粘包問(wèn)題的解決策略

由于底層的 TCP 無(wú)法理解上層業(yè)務(wù)數(shù)據(jù),所以在底層是無(wú)法保證數(shù)據(jù)包不被拆分和重組的?, 這個(gè)問(wèn)題只能通過(guò)上層的應(yīng)用協(xié)議棧設(shè)計(jì)來(lái)解決,根據(jù)業(yè)界主流的協(xié)議的解決方案, 可以歸納如下:

消息定長(zhǎng), 例如每個(gè)報(bào)文的大小固定長(zhǎng)度200字節(jié),如果不夠,空位補(bǔ)齊空格;

在包尾部添加回車換行符進(jìn)行分割, 例如 FTP 協(xié)議;

將消息分為消息頭和消息體,消息頭中包含表示消息總長(zhǎng)度(或者消息具體長(zhǎng)度)的字段,通常設(shè)計(jì)思路為消息頭的第一個(gè)字段使用 int32 來(lái)表示消息的總長(zhǎng)度;

更復(fù)雜的應(yīng)用協(xié)議層;

?一. LineBasedFrameDecoder 與?StringDecoder

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

LineBasedFrameDecoder?的工作原理是它依次遍歷 ByteBuf 中得可讀字節(jié),判斷看是否有 '\n' 或者 ?'\r\n' , ?如果有,就以此位置為結(jié)束位置,從可讀索引到結(jié)束位置區(qū)間的字節(jié)組成一行.他是以換行符為結(jié)束標(biāo)志的解碼器.支持?jǐn)y帶結(jié)束符或者不攜帶結(jié)束符兩種編碼方式,同時(shí)支持配置單行最大長(zhǎng)度 . 如果連續(xù)讀取到最大長(zhǎng)度后仍然沒(méi)有發(fā)現(xiàn)換行符,則拋出異常,同時(shí)忽略掉之前讀到的異常碼流. StringDecoder 的功能非常簡(jiǎn)單,就是將收到的對(duì)象轉(zhuǎn)換成字符串,然后繼續(xù)調(diào)用后面的handler? LineBasedFrameDecoder + StringDecoder ?組合就是按換行切換的文本解碼器, 它被設(shè)計(jì)來(lái)用于支持 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) {// 創(chuàng)建兩個(gè)NioEventLoopGroup 實(shí)例,NioEventLoopGroup// 是一個(gè)線程組,它包含一組NIO線程,專門用于處理網(wǎng)絡(luò)事件的處理,實(shí)際上他們就是Reactor 線程組// 這里創(chuàng)建兩個(gè)的原因是一個(gè)用于服務(wù)端接收用戶的鏈接,另一個(gè)用于進(jìn)行SocketChannel的網(wǎng)絡(luò)讀寫EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {// 創(chuàng)建一個(gè) ServerBootstrap ,它是netty用于NIO服務(wù)端的輔助啟動(dòng)類,目的是降低服務(wù)端的開發(fā)復(fù)雜度.ServerBootstrap bootstrap = new ServerBootstrap();// 設(shè)定 服務(wù)端接收用戶請(qǐng)求的線程組和用于進(jìn)行SocketChannel網(wǎng)絡(luò)讀寫的線程組 bootstrap.group(bossGroup, workerGroup);// 設(shè)置創(chuàng)建的 channel 類型bootstrap.channel(NioServerSocketChannel.class);// 配置 NioServerSocketChannel 的 tcp 參數(shù), BACKLOG 的大小bootstrap.option(ChannelOption.SO_BACKLOG, 1024);// 綁定io處理類（childChannelHandler）.他的作用類似于 reactor 模式中的 handler// 類,主要用于處理網(wǎng)絡(luò) I/O 事件,例如對(duì)記錄日志,對(duì)消息進(jìn)行解碼等.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());}});// 綁定端口,隨后調(diào)用它的同步阻塞方法 sync 等等綁定操作成功,完成之后 Netty 會(huì)返回一個(gè) ChannelFuture// 它的功能類似于的 Future,主要用于異步操作的通知回調(diào).ChannelFuture channelFuture = bootstrap.bind(port).sync();// 等待服務(wù)端監(jiān)聽端口關(guān)閉,調(diào)用 sync 方法進(jìn)行阻塞,等待服務(wù)端鏈路關(guān)閉之后 main 函數(shù)才退出. channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {// 優(yōu)雅的退出,釋放線程池資源 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();// 當(dāng)出現(xiàn)異常時(shí),釋放資源. 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());}});// 發(fā)起異步鏈接操作ChannelFuture future = bootstrap.connect(host, port).sync();// 等待客戶端鏈路關(guān)閉 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();// 當(dāng)出現(xiàn)異常時(shí),釋放資源. 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 定長(zhǎng)的分割器.? ch.pipeline().addLast(new FixedLengthFrameDecoder(1024));

?

? 以上內(nèi)容出自 : <Netty ?權(quán)威指南>?

轉(zhuǎn)載于:https://www.cnblogs.com/mjorcen/p/4539205.html

總結(jié)

以上是生活随笔為你收集整理的netty 粘包的解决策略的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。