1 产生原因

TCP是面向连接的，面向流的，提供高可靠性服务。收发两端（客户端和服务端）都要有一一成对的socket，因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化算法（Nagle算法），将许多数据量小且间隔小的数据，合并成了一个大数据块，然后进行封包，这样做虽然提升了效率，但是接收端就难于分辨出完成的数据包，因此面向流的通信是无消息保护边界的。

因为TCP无消息保护边界，需要在接收端处理消息边界问题，也就是所说的粘包和拆包问题。如下：

2 演示代码（TCP产生粘包|拆包）

客户端向服务端发送10条消息，看服务端如何接受。

服务端Server：

package com.liubujun.tcp;import com.liubujun.netty.NettyServerHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;/*** @Author: liubujun* @Date: 2023/2/26 13:35*/public class MyServer {public static void main(String[] args) throws Exception {//1.创建2个线程组bossGroup和workerGroup//2  bossGroup只是处理连接请求，workerGroup真正的和客户端进行业务处理//3 两个都是无限循环NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workerGroup = new NioEventLoopGroup();try {//创建服务器端的启动对象，配置参数ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup,workerGroup)//设置两个线程组.channel(NioServerSocketChannel.class) //使用nioSocketChannel作为服务器的通道实现.childHandler(new MyServerInitializer());//给workerGroup的EventLoop对应的管道设置处理器//绑定一个端口并且同步，生成了一个ChannelFuture对象//启动服务器（并绑定端口）ChannelFuture cf = bootstrap.bind(7000).sync();//对关联通道进行监听cf.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

服务端Handle：

package com.liubujun.tcp;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;import java.nio.charset.Charset;
import java.util.UUID;/*** @Author: liubujun* @Date: 2023/2/26 13:50*/public class MyServerHandle extends SimpleChannelInboundHandler<ByteBuf> {private int count;@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {byte[] buffer = new byte[msg.readableBytes()];msg.readBytes(buffer);//将buffer转成字符串String message = new String(buffer, Charset.forName("utf-8"));System.out.println("服务器端接收到数据"+message);System.out.println("服务器接收消息量="+(++this.count));//服务器回送数据给客户端，回送一个随机idByteBuf responseByteBuf = Unpooled.copiedBuffer(UUID.randomUUID().toString(), Charset.forName("utf-8"));ctx.writeAndFlush(responseByteBuf);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

服务端Initializer：

package com.liubujun.tcp;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;/*** @Author: liubujun* @Date: 2023/2/26 13:49*/public class MyServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyServerHandle());}
}

客户端：

package com.liubujun.tcp;import com.liubujun.netty.NettyClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;/*** @Author: liubujun* @Date: 2023/2/26 13:34*/public class MyClient {public static void main(String[] args) throws Exception{//客户端需要一个循环组EventLoopGroup group = new NioEventLoopGroup();try {//创建客户端的启动对象//注意客户端使用的是Bootstrap不是ServerBootstrapBootstrap bootstrap = new Bootstrap();bootstrap.group(group) //设置线程组.channel(NioSocketChannel.class) //设置客户端通道的实现类.handler(new MyClientInitializer());System.out.println("客户端 ok ...");//启动客户端连接服务器 ChannelFuture涉及到netty的异步模型ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync();//给关闭通道进行监听channelFuture.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
}

客户端Handle：

package com.liubujun.tcp;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;import java.nio.charset.Charset;/*** @Author: liubujun* @Date: 2023/2/26 13:40*/public class MyClientHandle extends SimpleChannelInboundHandler<ByteBuf> {private int count;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送十条数据hello，serverfor (int i = 0; i < 10; ++i) {ByteBuf buffer = Unpooled.copiedBuffer("hello,server" + i, Charset.forName("utf-8"));ctx.writeAndFlush(buffer);}}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {byte[] buffer =  new byte[msg.readableBytes()];msg.readBytes(buffer);String message = new String(buffer, Charset.forName("utf-8"));System.out.println("客户端接收消息="+message);System.out.println("客户端接收消息数量="+(++this.count));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

客户端Initializer：

package com.liubujun.tcp;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;/*** @Author: liubujun* @Date: 2023/2/26 13:37*/public class MyClientInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyClientHandle());}
}

启动服务端和客户端：

服务端控制台：一次全部接受

客户端控制台：只接受一次

发现进行了粘包

再次启动一个客户端：

服务端控制台：

客户端控制台：一次接受

发现进行了拆包。

3 TCP粘包|拆包解决方案

使用自定义协议+编码器解决

关键是解决服务器端每次读取数据长度的问题，这个问题解决，就不会出现服务器多读或少读数据的问题，从而避免TCP的拆包和粘包。

演示案例：

客户端发送5个Message对象，客户端每次发送一个message对象，服务器端就会每次接受一个message，分5次进行解码，每次读取到一个message，会回复一个message对象给客户端。

数据包：

//协议包
public class MessageProtocol {private int len;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;}
}

客户端代码：

public class MyClient {public static void main(String[] args) throws Exception{//客户端需要一个循环组EventLoopGroup group = new NioEventLoopGroup();try {//创建客户端的启动对象//注意客户端使用的是Bootstrap不是ServerBootstrapBootstrap bootstrap = new Bootstrap();bootstrap.group(group) //设置线程组.channel(NioSocketChannel.class) //设置客户端通道的实现类.handler(new MyClientInitializer());System.out.println("客户端 ok ...");//启动客户端连接服务器 ChannelFuture涉及到netty的异步模型ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync();//给关闭通道进行监听channelFuture.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
}

客户端处理器：

public class MyClientHandle extends SimpleChannelInboundHandler<MessageProtocol> {private int count;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送十条数据今天天气冷，来吃火锅for (int i = 0; i < 5; ++i) {String msg = "今天天气冷，来吃火锅";byte[] content = msg.getBytes(Charset.forName("utf-8"));int length = content.length;MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);ctx.writeAndFlush(messageProtocol);}}@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,Charset.forName("utf-8")));System.out.println("客户端接收消息数量="+(++this.count));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常消息:"+cause.getMessage());ctx.close();}
}

客户端初始化：

public class MyClientInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageEncoder()); //加入编码器pipeline.addLast(new MyClientHandle());pipeline.addLast(new MyMessageDecoder());}
}

编码器：

public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {@Overrideprotected void encode(ChannelHandlerContext channelHandlerContext, MessageProtocol msg, ByteBuf out) throws Exception {System.out.println("MyMessageEncoder encoder 方法被调用");out.writeInt(msg.getLen());out.writeBytes(msg.getContent());}
}

解码器：

public class MyMessageDecoder extends ReplayingDecoder<Void> {@Overrideprotected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> list) throws Exception {System.out.println("MyMessageDecoder decoder 被调用");//需要将得到的二进制字节码->MessageProtocol 数据包(对象)int length = in.readInt();byte[] content = new byte[length];in.readBytes(content);//封装成MessageProtocol 对象，放入out，传递下一个handle业务处理MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);list.add(messageProtocol);}
}

服务端：

public class MyServer {public static void main(String[] args) throws Exception {//1.创建2个线程组bossGroup和workerGroup//2  bossGroup只是处理连接请求，workerGroup真正的和客户端进行业务处理//3 两个都是无限循环NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workerGroup = new NioEventLoopGroup();try {//创建服务器端的启动对象，配置参数ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup,workerGroup)//设置两个线程组.channel(NioServerSocketChannel.class) //使用nioSocketChannel作为服务器的通道实现.childHandler(new MyServerInitializer());//给workerGroup的EventLoop对应的管道设置处理器//绑定一个端口并且同步，生成了一个ChannelFuture对象//启动服务器（并绑定端口）ChannelFuture cf = bootstrap.bind(7000).sync();//对关联通道进行监听cf.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

服务端处理器：

public class MyServerHandle extends SimpleChannelInboundHandler<MessageProtocol> {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,Charset.forName("utf-8")));System.out.println("服务器接收到消息包数量:"+(++this.count));//回复消息String responseContent = UUID.randomUUID().toString();int length = responseContent.getBytes("utf-8").length;byte[] bytes = responseContent.getBytes("utf-8");//构建一个协议包MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(bytes);ctx.writeAndFlush(messageProtocol);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}

服务端初始化：

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageDecoder()); //因为要接收消息加入解码器pipeline.addLast(new MyServerHandle());pipeline.addLast(new MyMessageEncoder());//因为要恢复消息加入编码器}
}

启动服务端和客户端：

服务端控制台输出：