本文来源:《Netty权威指南》
TCP 粘包/拆包 TCP 是一个“流”协议,没有分界线。 TCP底层并不了解上层业务数据的具体含义,它会根据TCP缓冲区的实际情况进行包的划分。 粘包/拆包现象: 如下图,客户端发送两个数据包D1和D2到服务器,会出现以下4种情况: (1)服务端分两次读取,获得两个独立的数据包,分别D1和D2,这种情况下没有粘包/拆包问题。 (2)服务端一共就读到了一个数据包,这个数据包是D1和D2的完整信息,这样D1和D2粘合在一起,因为服务端不知道第一条消息从哪儿结束和第二条消息从哪儿开始,所以发生了TCP粘包。 (3)服务端分两次读取到了两个数据包,第一次读取到了完整的D1包和D2包的前半部分D2_1,第二次读取到了D2包的剩余部分D2_2,这被称为TCP拆包。 (4)服务端分两次读取到了两个数据包,第一次读取到了D1包的前半部分D1_1,第二次读取到了D1包的剩余部分D1_2和D2包的整包,这也被称为TCP拆包。 粘包/拆包会导致半包读写问题!! 发生粘包/拆包的主要原因: (1)应用程序写入的数据大于套接字缓冲区大小,这将会发生拆包。 (2)应用程序写入数据小于套接字缓冲区大小,网卡将应用多次写入的数据发送到网络上,这将会发生粘包。 (3)进行MSS(最大报文长度)大小的TCP分段,当TCP报文长度-TCP头部长度>MSS的时候将发生拆包。 (4)接收方不及时读取套接字缓冲区数据,这将发生粘包。 (5)…… 粘包/拆包问题的解决策略 通过上层的应用协议栈来解决。 (1)消息定长。例如每个报文的大小为固定长度200字节,如果不够,空位补空格。 (2)在包尾增加回车换行符进行分割,例如FTP协议。 (3)设置消息边界,服务端从网络流中按消息边界分离出消息内容。 (3)将消息分为消息头和消息体,消息头中包含表示消息总长度(或者消息体长度)的字段。 (4)更复杂的应用层协议。 下面看看,如何利用Netty提供的半包解码器来解决TCP粘包/拆包问题。 一、未考虑TCP粘包问题的程序 未进行粘包/拆包处理的代码: package com.tao.netty.netty.ch4; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; public class TimeServer { public void bind(int port) throws Exception { //配置服务器端的NIO线程组 //bossGroup用于服务端接收客户端的连接 //workerGroup用于网络读写 EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // TODO Auto-generated method stub ch.pipeline().addLast(new TimeServerHandler()); } }); //绑定端口,同步等待成功 ChannelFuture f = b.bind(port).sync(); //等待服务端监听端口关闭 f.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { //优雅退出,释放线程池资源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; new TimeServer().bind(port); } } package com.tao.netty.netty.ch4; import java.util.Date; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; public class TimeServerHandler extends ChannelInboundHandlerAdapter { private int counter; @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // TODO Auto-generated method stub ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req);//读数据到req中 String body = new String(req, "UTF-8").substring(0, req.length - System.getProperty("line.separator").length()); System.out.println("服务器收到客户端发送的命令:" + body + ", 计数器counter:" + (++counter)); String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "无效的命令"; currentTime = currentTime + System.getProperty("line.separator"); ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes()); ctx.writeAndFlush(resp); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { // TODO Auto-generated method stub ctx.close(); } } package com.tao.netty.netty.ch4; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; public class TimeClient { public void connect(String host, int port) throws Exception { //配置客户端NIO线程组 EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new TimeClientHandler()); } }); //发起异步连接请求 ChannelFuture f = b.connect(host, port).sync(); f.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { group.shutdownGracefully(); } } public static void main(String[] args) { int port = 8080; try { new TimeClient().connect("127.0.0.1", port); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } } } package com.tao.netty.netty.ch4; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; public class TimeClientHandler extends ChannelInboundHandlerAdapter { private int counter; private byte[] req; public TimeClientHandler() { req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes(); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { // TODO Auto-generated method stub ByteBuf message = null; //发送100次请求 for(int i = 0; i < 100; i++) { message = Unpooled.buffer(req.length); message.writeBytes(req); ctx.writeAndFlush(message); } } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // TODO Auto-generated method stub ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "UTF-8"); System.out.println("当前服务器时间:" + body + ", 计数器counter:" + counter); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println(cause.getMessage()); ctx.close(); } } 本意是让客户端发送100条请求消息,服务器发回100条应答消息。但由于发生了TCP粘包,客户端发送的消息数不足100条,并且服务器无法解析粘包中的命令。 --------------------------------------------------------------------------------------------------------------------------------- Netty中有两个抽象类: 1、ByteToMessageDecoder 2、MessageToMessageDecoder 这两个组件都实现了ChannelInboundHandler接口,这说明这两个组件都是用来解码网络上过来的数据的。而他们的顺序一般是ByteToMessageDecoder位于head channel handler的后面,MessageToMessageDecoder位于ByteToMessageDecoder的后面。Netty中,涉及到粘包、拆包的逻辑主要在ByteToMessageDecoder及其实现中。 当然,使用 ByteToMessageDecoder,用户需要自己去实现处理粘包、拆包的逻辑,这还是有一定难度的。 Netty已经提供了一些基于不同处理粘包、拆包规则的实现: DelimiterBasedFrameDecoder 是基于消息边界方式进行粘包拆包处理的。 FixedLengthFrameDecoder 是基于固定长度消息进行粘包拆包处理的。 LengthFieldBasedFrameDecoder 是基于消息头指定消息长度进行粘包拆包处理的。 LineBasedFrameDecoder 是基于换行符来进行消息粘包拆包处理的。 用户可以自行选择规则然后使用Netty提供的对应的Decoder来进行具有粘包、拆包处理功能的网络应用开发。 --------------------------------------------------------------------------------------------------------------------------------- 下面展示使用LineBasedFrameDecoder 解码器 + StringDecoder解码器 来解决粘包拆包问题。 LineBasedFrameDecoder 工作原理: StringDecoder 工作原理: 二、考虑TCP粘包问题的程序 支持TCP粘包/拆包的程序: 需要在原来的TimeServerHandler之前添加两个解码器:LineBasedFrameDecoder 和 StringDecoder package com.tao.netty.netty.ch4.code2; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.LineBasedFrameDecoder; import io.netty.handler.codec.string.StringDecoder; public class TimeServer { public void bind(int port) throws Exception { //配置服务器端的NIO线程组 //bossGroup用于服务端接收客户端的连接 //workerGroup用于网络读写 EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new LineBasedFrameDecoder(1024)); //添加LineBasedFrameDecoder解码器 ch.pipeline().addLast(new StringDecoder()); //添加StringDecoder解码器 ch.pipeline().addLast(new TimeServerHandler()); } }); //绑定端口,同步等待成功 ChannelFuture f = b.bind(port).sync(); //等待服务端监听端口关闭 f.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { //优雅退出,释放线程池资源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; new TimeServer().bind(port); } } package com.tao.netty.netty.ch4.code2; import java.util.Date; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; public class TimeServerHandler extends ChannelInboundHandlerAdapter { private int counter; @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //这里的msg就是删除了回车换行符之后的请求消息 String body = (String) msg; System.out.println("服务器收到客户端发送的命令:" + body + ", 计数器counter:" + (++counter)); String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "无效的命令"; currentTime = currentTime + System.getProperty("line.separator"); ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes()); ctx.writeAndFlush(resp); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { // TODO Auto-generated method stub ctx.close(); } } package com.tao.netty.netty.ch4.code2; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.LineBasedFrameDecoder; import io.netty.handler.codec.string.StringDecoder; public class TimeClient { public void connect(String host, int port) throws Exception { //配置客户端NIO线程组 EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new LineBasedFrameDecoder(1024)); ch.pipeline().addLast(new StringDecoder()); ch.pipeline().addLast(new TimeClientHandler()); } }); //发起异步连接请求 ChannelFuture f = b.connect(host, port).sync(); f.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); } finally { group.shutdownGracefully(); } } public static void main(String[] args) { int port = 8080; try { new TimeClient().connect("127.0.0.1", port); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); } } } package com.tao.netty.netty.ch4.code2; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; public class TimeClientHandler extends ChannelInboundHandlerAdapter { private int counter; private byte[] req; public TimeClientHandler() { req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes(); } //连接建立时触发 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { // TODO Auto-generated method stub ByteBuf message = null; //发送100次请求 for(int i = 0; i < 100; i++) { message = Unpooled.buffer(req.length); message.writeBytes(req); ctx.writeAndFlush(message); } } //收到数据时触发 @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // TODO Auto-generated method stub String body = (String) msg; System.out.println("当前服务器时间:" + body + ", 计数器counter:" + (++counter)); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println(cause.getMessage()); ctx.close(); } } System.getProperty("line.separator")的作用是获得换行符,与平台无关。