本文来源：《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")的作用是获得换行符，与平台无关。