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());//因为要恢复消息加入编码器}
}
启动服务端和客户端:
服务端控制台输出:
客户端控制台输出:
可以发现,发送5条消息,服务端接收5次。