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。
相关内容
热门资讯
原创 就...
【军武次位面】作者：乐乐 日前，美国“Military Watch”网站报道称，中国海军一艘“基洛”...
输球又输点！阿森纳赛后点球3-...
在刚刚结束的季前友谊赛中，阿森纳以2-3不敌比利亚雷亚尔，随后在点球大战中以3-4失利，令人意外的是...
8月1日起乌鲁木齐天山国际机场...
2025年8月1日起，新疆机场集团乌鲁木齐天山国际机场将迎来一项关键服务升级：所有国内出港航班值机手...
“我就在这儿坐着怎么了”，火车...
安全乘车，文明出行，是每一位公民应尽的责任和义务。近日，旅客李某持无座车票强占其他旅客座位，经乘警多...
黑龙江省制定出台20条政策措施...
近日，黑龙江制定出台支持高端智能农机装备产业高质量发展20条政策措施。旨在引导产学研用等各方用好国家...
债券利息收入增值税新规落地在即...
债券利息收入税收新规实施前夕，政策性银行密集发行金融债。 8月5日，中国债券信息网披露的信息显示，中...
静乐县公安局征集“六霸”及殡葬...
为深入开展群众身边不正之风和腐败问题集中整治，严厉打击“六霸”及殡葬等领域涉民生违法犯罪，现向社会各...
原创 欧...
欧洲媒体在8月5日的报道中提到，美国与欧洲似乎达成一致，准备联合打压俄罗斯石油的主要买家——中国和印...
普京与美特使聊了3小时之后，特...
来源：视觉中国 俄罗斯总统普京与美国特使威特科夫的会晤在持续近3小时后结束。 据新华社报道，俄总统助...