Netty解决 TCP 粘包拆包的方法
作者：Java技术栈 发布时间：2023-03-09 14:39:06
什么是粘包/拆包
一般所谓的TCP粘包是在一次接收数据不能完全地体现一个完整的消息数据。TCP通讯为何存在粘包呢？主要原因是TCP是以流的方式来处理数据，再加上网络上MTU的往往小于在应用处理的消息数据，所以就会引发一次接收的数据无法满足消息的需要，导致粘包的存在。处理粘包的唯一方法就是制定应用层的数据通讯协议，通过协议来规范现有接收的数据是否满足消息数据的需要。
我们都知道TCP是基于字节流的传输协议。
那么数据在通信层传播其实就像河水一样并没有明显的分界线，而数据具体表示什么意思什么地方有句号什么地方有分号这个对于TCP底层来说并不清楚。应用层向TCP层发送用于网间传输的、用8位字节表示的数据流，然后TCP把数据流分区成适当长度的报文段，之后TCP把结果包传给IP层，由它来通过网络将包传送给接收端实体的TCP层。
所以对于这个数据拆分成大包小包的问题就是我们今天要讲的粘包和拆包的问题。
1、TCP粘包拆包问题说明
粘包和拆包这两个概念估计大家还不清楚，通过下面这张图我们来分析一下：
假设客户端分别发送两个数据包D1,D2个服务端，但是发送过程中数据是何种形式进行传播这个并不清楚，分别有下列4种情况：
服务端一次接受到了D1和D2两个数据包，两个包粘在一起，称为粘包；
服务端分两次读取到数据包D1和D2，没有发生粘包和拆包；
服务端分两次读到了数据包，第一次读到了D1和D2的部分内容，第二次读到了D2的剩下部分，这个称为拆包；
服务器分三次读到了数据部分，第一次读到了D1包，第二次读到了D2包的部分内容，第三次读到了D2包的剩下内容。
2、TCP粘包产生原因
我们知道在TCP协议中，应用数据分割成TCP认为最适合发送的数据块，这部分是通过“MSS”（最大数据包长度）选项来控制的，通常这种机制也被称为一种协商机制，MSS规定了TCP传往另一端的最大数据块的长度。这个值TCP协议在实现的时候往往用MTU值代替（需要减去IP数据包包头的大小20Bytes和TCP数据段的包头20Bytes）所以往往MSS为1460。通讯双方会根据双方提供的MSS值得最小值确定为这次连接的最大MSS值。
tcp为提高性能，发送端会将需要发送的数据发送到缓冲区，等待缓冲区满了之后，再将缓冲中的数据发送到接收方。同理，接收方也有缓冲区这样的机制，来接收数据。
发生粘包拆包的原因主要有以下这些：
应用程序写入数据的字节大小大于套接字发送缓冲区的大小将发生拆包；
进行MSS大小的TCP分段。MSS是TCP报文段中的数据字段的最大长度，当TCP报文长度-TCP头部长度>mss的时候将发生拆包；
应用程序写入数据小于套接字缓冲区大小，网卡将应用多次写入的数据发送到网络上,将发生粘包；
数据包大于MTU的时候将会进行切片。MTU即(Maxitum Transmission Unit) 最大传输单元,由于以太网传输电气方面的限制，每个以太网帧都有最小的大小64bytes最大不能超过1518bytes,刨去以太网帧的帧头14Bytes和帧尾CRC校验部分4Bytes,那么剩下承载上层协议的地方也就是Data域最大就只能有1500Bytes这个值我们就把它称之为MTU。这个就是网络层协议非常关心的地方，因为网络层协议比如IP协议会根据这个值来决定是否把上层传下来的数据进行分片。
3、如何解决TCP粘包拆包
我们知道tcp是 * 的数据流，且协议本身无法避免粘包，拆包的发生，那我们只能在应用层数据协议上，加以控制。通常在制定传输数据时，可以使用如下方法：
设置定长消息，服务端每次读取既定长度的内容作为一条完整消息；
使用带消息头的协议、消息头存储消息开始标识及消息长度信息，服务端获取消息头的时候解析出消息长度，然后向后读取该长度的内容；
设置消息边界，服务端从网络流中按消息边界分离出消息内容。比如在消息末尾加上换行符用以区分消息结束。
当然应用层还有更多复杂的方式可以解决这个问题，这个就属于网络层的问题了，我们还是用java提供的方式来解决这个问题。我们先看一个例子看看粘包是如何发生的。
服务端：
public class HelloWordServer {
private int port;
public HelloWordServer(int port) {
this.port = port;
}
public void start(){
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ServerChannelInitializer());
try {
ChannelFuture future = server.bind(port).sync();
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
HelloWordServer server = new HelloWordServer(7788);
server.start();
}
}
服务端Initializer：
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
// 字符串解码 和 编码
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 自己的逻辑Handler
pipeline.addLast("handler", new HelloWordServerHandler());
}
}
服务端handler：
public class HelloWordServerHandler extends ChannelInboundHandlerAdapter {
private int counter;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String)msg;
System.out.println("server receive order : " + body + ";the counter is: " + ++counter);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
}
客户端：
public class HelloWorldClient {
private int port;
private String address;
public HelloWorldClient(int port,String address) {
this.port = port;
this.address = address;
}
public void start(){
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ClientChannelInitializer());
try {
ChannelFuture future = bootstrap.connect(address,port).sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
HelloWorldClient client = new HelloWorldClient(7788,"127.0.0.1");
client.start();
}
}
客户端Initializer：
public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 客户端的逻辑
pipeline.addLast("handler", new HelloWorldClientHandler());
}
}
客户端handler：
public class HelloWorldClientHandler extends ChannelInboundHandlerAdapter {
private byte[] req;
private int counter;
public BaseClientHandler() {
req = ("Unless required by applicable law or agreed to in writing, software\n" +
" distributed under the License is distributed on an \"AS IS\" BASIS,\n" +
" WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" +
" See the License for the specific language governing permissions and\n" +
" limitations under the License.This connector uses the BIO implementation that requires the JSSE\n" +
" style configuration. When using the APR/native implementation, the\n" +
" penSSL style configuration is required as described in the APR/native\n" +
" documentation.An Engine represents the entry point (within Catalina) that processes\n" +
" every request. The Engine implementation for Tomcat stand alone\n" +
" analyzes the HTTP headers included with the request, and passes them\n" +
" on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software\n" +
"# distributed under the License is distributed on an \"AS IS\" BASIS,\n" +
"# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n" +
"# See the License for the specific language governing permissions and\n" +
"# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log\n" +
"# each component that extends LifecycleBase changing state:\n" +
"#org.apache.catalina.util.LifecycleBase.level = FINE"
).getBytes();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message;
//将上面的所有字符串作为一个消息体发送出去
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String buf = (String)msg;
System.out.println("Now is : " + buf + " ; the counter is : "+ (++counter));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
运行客户端和服务端我们能看到：
我们看到这个长长的字符串被截成了2段发送，这就是发生了拆包的现象。同样粘包我们也很容易去模拟，我们把BaseClientHandler中的channelActive方法里面的：
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
这几行代码是把我们上面的一长串字符转成的byte数组写进流里发送出去，那么我们可以在这里把上面发送消息的这几行循环几遍这样发送的内容增多了就有可能在拆包的时候把上一条消息的一部分分配到下一条消息里面了，修改如下：
for (int i = 0; i < 3; i++) {
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
改完之后我们再运行一下，输出太长不好截图，我们在输出结果中能看到循环3次之后的消息服务端收到的就不是之前的完整的一条了，而是被拆分了4次发送。
对于上面出现的粘包和拆包的问题，Netty已有考虑，并且有实施的方案：LineBasedFrameDecoder。
我们重新改写一下ServerChannelInitializer：
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new LineBasedFrameDecoder(2048));
// 字符串解码 和 编码
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 自己的逻辑Handler
pipeline.addLast("handler", new BaseServerHandler());
}
}
新增：pipeline.addLast(new LineBasedFrameDecoder(2048))。同时，我们还得对上面发送的消息进行改造BaseClientHandler：
public class BaseClientHandler extends ChannelInboundHandlerAdapter {
private byte[] req;
private int counter;
req = ("Unless required by applicable dfslaw or agreed to in writing, software" +
" distributed under the License is distributed on an \"AS IS\" BASIS," +
" WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied." +
" See the License for the specific language governing permissions and" +
" limitations under the License.This connector uses the BIO implementation that requires the JSSE" +
" style configuration. When using the APR/native implementation, the" +
" penSSL style configuration is required as described in the APR/native" +
" documentation.An Engine represents the entry point (within Catalina) that processes" +
" every request. The Engine implementation for Tomcat stand alone" +
" analyzes the HTTP headers included with the request, and passes them" +
" on to the appropriate Host (virtual host)# Unless required by applicable law or agreed to in writing, software" +
"# distributed under the License is distributed on an \"AS IS\" BASIS," +
"# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied." +
"# See the License for the specific language governing permissions and" +
"# limitations under the License.# For example, set the org.apache.catalina.util.LifecycleBase logger to log" +
"# each component that extends LifecycleBase changing state:" +
"#org.apache.catalina.util.LifecycleBase.level = FINE\n"
).getBytes();
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message;
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String buf = (String)msg;
System.out.println("Now is : " + buf + " ; the counter is : "+ (++counter));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
去掉所有的”\n”,只保留字符串末尾的这一个。原因稍后再说。channelActive方法中我们不必再用循环多次发送消息了，只发送一次就好(第一个例子中发送一次的时候是发生了拆包的)，然后我们再次运行，大家会看到这么长一串字符只发送了一串就发送完毕。程序输出我就不截图了。下面来解释一下LineBasedFrameDecoder。
LineBasedFrameDecoder的工作原理是它依次遍历ByteBuf 中的可读字节，判断看是否有”\n” 或者” \r\n”，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器。支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。如果连续读取到最大长度后仍然没有发现换行符，就会抛出异常，同时忽略掉之前读到的异常码流。这个对于我们确定消息最大长度的应用场景还是很有帮助。
对于上面的判断看是否有”\n” 或者” \r\n”以此作为结束的标志我们可能回想，要是没有”\n” 或者” \r\n”那还有什么别的方式可以判断消息是否结束呢。别担心，Netty对于此已经有考虑，还有别的解码器可以帮助我们解决问题，
来源：https://www.cnblogs.com/javastack/archive/2021/07/05/14972520.html
0
投稿猜你喜欢
- 上一篇我们详解了setttings.xml的配置项，里面的配置项基本都和仓库有关系，我们使用maven更多的也是要从仓库下载jar包，然后也
- 1.我做的是一个动态表格，就是在输入框里每输入一次数据并点击“添加”按钮，表格中就会新增一行记录。<table id="st
- 一、延迟加载：LazyLoading使用延迟加载，关联的实体必须标注为virtual。本例是标注Destination类里的Lodgings
前言什么是mybatis二级缓存？二级缓存是多个sqlsession共享的，其作用域是mapper的同一个namespace。即，在不同的s- 算法简介迪杰斯特拉算法(Dijkstra)是由荷兰计算机科学迪家迪杰斯特拉于1959年提出的，因此又叫狄克斯特拉算法。是从一个顶点到其余各顶
- 此处网上最多的做法是需要修改tomcat的参数配置大致如下：<Connector port="8080" prot
- 其实本没有没打算写这篇的，但还是要写一下写这篇博客的起因是因为，现在呆着的这家公司居然没有统一的API返回格式?，询问主管他居然告诉我用HT
- JNI中的java接口使用项目需求，需要在c++函数中监听相应的状态，并在java端进行一些列的处理。这个需要在JNI中写一个subscri
- 实践过程效果代码public partial class Form1 : Form{ public Form1()
- 在做在线编程题目的时候，需要了解一下数据的输入格式。这样可以对数据处理有比较好的把握，不需要把太多的时间放在这个上面，注重主要的算法逻辑即可
- spring xml文件打开没有namespace等操作选项第一步查看自己安装的eclipse的正确的版本号：打开eclipse，点击hel
- 基本思想：归并（Merge）排序法是将两个（或两个以上）有序表合并成一个新的有序表，即把待排序序列分为若干个子序列，每个子序列是有序的。然后
- private void btnCreate_Click(object sender, EventArgs e) ...{ int hWnd
- ReentrantLock锁ReentrantLock是Java中常用的锁，属于乐观锁类型，多线程并 * 况下。能保证共享数据安全性，线程间有
- 部署到webapps目录启动本文使用的Spring版本为Spring6，SpringBoot版本为3，JDK为17，可能会和之前有细微不同，
- java String的深入理解一、Java内存模型 按照官方的说法：Java 虚拟机具有一个堆，堆是运行时数据区域，所有类实例和
- 本文介绍了JAVA中实现原生的 socket 通信机制原理，分享给大家，具体如下：当前环境jdk == 1.8知识点socket 的连接处理
- 最近在做学校的课程设计，java编程需要用到对话框弹出，第一反应是js中的alert和confirm，java的话瞬间懵，查阅学习总结如下，
- 好久就想着好好搭建一个ssm框架，自己以后用也方便吧，但是最近的事真的是很多，很多事情都没有去干，有时候自己会怀疑一下人生自己该不该去做程序
- 一、简介约瑟夫问题（有时也称为约瑟夫斯置换，是一个出现在计算机科学和数学中的问题。在计算机编程的算法中，类似问题又称为约瑟夫环。又称“丢手绢
