Java基础之《netty（29）—自定义协议解决TCP粘包拆包》
创始人
2024-05-15 14:16:16
0次
一、TCP粘包和拆包解决方案
1、使用自定义协议 + 编解码器，来解决。
2、关键就是要解决，服务器端每次读取数据长度的问题。这个问题解决，就不会出现服务器多读或少读数据的问题，从而避免TCP粘包、拆包。
二、具体示例
1、要求客户端发送5个message对象，客户端每次发送一个message对象。
2、服务器端每次接收一个message，分5次进行解码，每读取到一个message，会回复一个message对象给客户端。
三、客户端发送给服务端
1、编解码器和POJO类
MessageProtocol.java
package netty.tcpStickPackage2;/*** 协议包对象类* @author user**/
public class MessageProtocol {//数据发送的长度private int len;//数据本体，一般是放在byte数组中private byte[] content;public int getLen() {return len;}public void setLen(int len) {this.len = len;}public byte[] getContent() {return content;}public void setContent(byte[] content) {this.content = content;}}MyMessageDecoder.java
package netty.tcpStickPackage2;import java.util.List;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;public class MyMessageDecoder extends ReplayingDecoder {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List MyMessageEncoder.java
package netty.tcpStickPackage2;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;public class MyMessageEncoder extends MessageToByteEncoder {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {System.out.println("MyMessageEncoder encode 方法被调用");out.writeInt(msg.getLen());out.writeBytes(msg.getContent());}} 2、服务端
NettyServer.java
package netty.tcpStickPackage2;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;public class NettyServer {public static void main(String[] args) {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup(8);try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.DEBUG)).childHandler(new NettyServerInitializer()); //自定义一个初始化类ChannelFuture cf = bootstrap.bind(7000).sync();cf.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}NettyServerInitializer.java
package netty.tcpStickPackage2;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class NettyServerInitializer extends ChannelInitializer {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//加入解码器pipeline.addLast(new MyMessageDecoder());//加入一个自定义handlerpipeline.addLast(new NettyChannelHandler());}} NettyChannelHandler.java
package netty.tcpStickPackage2;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;public class NettyChannelHandler extends SimpleChannelInboundHandler {private int count;@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {//接收到数据并处理int len = msg.getLen();byte[] content = msg.getContent();System.out.println("服务端接收到信息如下");System.out.println("长度：" + len);System.out.println("内容：" + new String(content, CharsetUtil.UTF_8));System.out.println("服务器接收到消息包数量：" + (++this.count));//回复消息}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.channel().close();}} 3、客户端
NettyClient.java
package netty.tcpStickPackage2;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;public class NettyClient {public static void main(String[] args) {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group) //设置线程组.channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.DEBUG)).handler(new NettyClientInitializer()); //自定义一个初始化对象ChannelFuture cf = bootstrap.connect("127.0.0.1", 7000).sync();cf.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {group.shutdownGracefully();}}
}NettyClientInitializer.java
package netty.tcpStickPackage2;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class NettyClientInitializer extends ChannelInitializer {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//加入编码器pipeline.addLast(new MyMessageEncoder());//加入一个自定义handlerpipeline.addLast(new NettyClientHandler());}} NettyClientHandler.java
package netty.tcpStickPackage2;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;public class NettyClientHandler extends SimpleChannelInboundHandler {//private int count;@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送5条数据for (int i=0; i<5; i++) {String msg = "今天天气冷，吃火锅";byte[] content = msg.getBytes(CharsetUtil.UTF_8);int length = content.length;//创建协议包MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);ctx.writeAndFlush(messageProtocol);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.channel().close();}} 四、执行结果
1、服务端
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：1
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：2
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：3
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：4
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：52、客户端
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用五、服务端回复消息给客户端
1、修改服务端代码
NettyChannelHandler.java
添加回复消息
package netty.tcpStickPackage2;import java.util.UUID;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;public class NettyChannelHandler extends SimpleChannelInboundHandler {private int count;@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {//接收到数据并处理int len = msg.getLen();byte[] content = msg.getContent();System.out.println("服务端接收到信息如下");System.out.println("长度：" + len);System.out.println("内容：" + new String(content, CharsetUtil.UTF_8));System.out.println("服务器接收到消息包数量：" + (++this.count));//回复消息String response = UUID.randomUUID().toString();byte[] responseContent = response.getBytes(CharsetUtil.UTF_8);int responseLen = responseContent.length;//构建一个协议包MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(responseLen);messageProtocol.setContent(responseContent);ctx.writeAndFlush(messageProtocol);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.channel().close();}} NettyServerInitializer.java
加入编码器
package netty.tcpStickPackage2;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class NettyServerInitializer extends ChannelInitializer {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//加入解码器pipeline.addLast(new MyMessageDecoder());//加入编码器pipeline.addLast(new MyMessageEncoder());//加入一个自定义handlerpipeline.addLast(new NettyChannelHandler());}} 2、修改客户端代码
NettyClientHandler.java
添加接收服务端信息channelRead0接口
package netty.tcpStickPackage2;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;public class NettyClientHandler extends SimpleChannelInboundHandler {private int count;@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {int len = msg.getLen();byte[] content = msg.getContent();System.out.println("客户端接收到消息如下");System.out.println("长度：" + len);System.out.println("内容：" + new String(content, CharsetUtil.UTF_8));System.out.println("客户端接收到消息包数量：" + (++this.count));}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送5条数据for (int i=0; i<5; i++) {String msg = "今天天气冷，吃火锅";byte[] content = msg.getBytes(CharsetUtil.UTF_8);int length = content.length;//创建协议包MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);ctx.writeAndFlush(messageProtocol);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.channel().close();}} NettyClientInitializer.java
添加解码器
package netty.tcpStickPackage2;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;public class NettyClientInitializer extends ChannelInitializer {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//加入编码器pipeline.addLast(new MyMessageEncoder());//加入解码器pipeline.addLast(new MyMessageDecoder());//加入一个自定义handlerpipeline.addLast(new NettyClientHandler());}} 六、执行结果
1、服务端
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：1
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：2
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：3
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：4
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用
服务端接收到信息如下
长度：27
内容：今天天气冷，吃火锅
服务器接收到消息包数量：5
MyMessageEncoder encode 方法被调用2、客户端
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用
客户端接收到消息如下
长度：36
内容：6a488e60-a67f-49ef-9783-cbcb048d82ea
客户端接收到消息包数量：1
MyMessageDecoder decode 被调用
客户端接收到消息如下
长度：36
内容：160ab489-09b9-4786-ac57-74b82aed7150
客户端接收到消息包数量：2
MyMessageDecoder decode 被调用
客户端接收到消息如下
长度：36
内容：d9fd2087-1d6f-474d-93d4-6af1c1ced736
客户端接收到消息包数量：3
MyMessageDecoder decode 被调用
客户端接收到消息如下
长度：36
内容：06b6b40e-6ab4-4374-9e5a-2fcaca8dbc4d
客户端接收到消息包数量：4
MyMessageDecoder decode 被调用
客户端接收到消息如下
长度：36
内容：40891ed6-a790-4660-962f-4e3ec9d3a99b
客户端接收到消息包数量：5七、其他
1、因为在encoder中先writeInt再writeByte，然后decoder中先readInt再readByte，这样每次接收数据的数据长度都是一个int+一个content的长度。
2、这个decode如果发生拆包的现象的时候会抛出一个Relay的Error，然后等待下一个请求进来，只有长度符合之后才会进行read，源码的位置是ReplayingDecoder的388行左右。
3、粘包就是通过定长来获取足够的字节码。拆包就是通过readBytes的时候需要的length和in的长度进行比较，短了就抛出error。
相关内容
热门资讯
集邦咨询：三地“芯”政策指向“...
6月18日，TrendForce集邦咨询发布了一份对广州、珠海、上海三地近期出台的集成电路产业新政的...
鲍威尔：货币政策仍需要“适度限...
美联储主席鲍威尔在新闻发布会上表示，鉴于通胀现状，仍需要货币政策对经济施加一些限制。目前的利率水平不...
李源祥：上海和香港应在监管、政...
友邦保险集团首席执行官兼总裁、日内瓦保险协会主席李源祥 东方网 图 6月18日，友邦保险集团首席执行...
哈梅内伊强硬发声，消息人士：他...
在美国总统特朗普发文要求伊朗“无条件投降”，甚至和以色列总理内塔尼亚胡先后对伊朗最高领袖哈梅内伊发出...
女子4年前借贷900万元，拿下...
借用900万元经营贷，“打新”总价近3000万元的豪宅，不料交房快两年了，始终没有办下不动产权证。 ...
鲍威尔：当前的政策立场做好灵活...
美联储主席鲍威尔表示，当前的政策立场做好灵活应对准备。
又有美企直诉至美国最高法院 特...
[ 根据美国最高法院公布的程序，对于不满意下级法院裁决的一方，可向美国最高法院提出请愿，请求法院审理...
特朗普回应哈梅内伊拒绝投降
据美联社18日消息，伊朗最高领袖哈梅内伊周三早些时候警告称，美国若对伊朗发动打击，“将自食其果，遭受...
全力押注枪手罗体：约克雷斯告知...
直播吧06月18日讯 据意媒《罗马体育报》报道，约克雷斯不会加盟尤文图斯，球员及其经纪人目的地是北伦...
两架伊朗政府飞机已经离开伊朗，...
格隆汇6月18日｜两架伊朗政府飞机已经离开伊朗，飞往阿曼；阿曼作为调解人在美伊谈判中发挥了关键作用。...