说明：

粘包：

现象：
发送abc、def，接收abcdef
原因：

  1. 应用层：接收方ByteBuf设置太大(Netty默认1024)
  2. 滑动窗口：假设发送方256 bytes示-个完整报文,但由于接收方处理不及时且窗口大小足够大，这256 bytes字节就会缓冲在接收方的滑动窗口中，当滑动窗口中缓冲了多个报文就会粘包
  3. Netty使用了Nagle算法将多次间隔较小且数据量小的数据合并成一个大的数据块进行封包发送

拆包（半包）：

现象：
发送abcdef，接收abc、def
原因：

  1. 应用层：接收方ByteBuf小于实际发送数据量
  2. 滑动窗口：假设接收方的窗口只剩了128 bytes,发送方的报文大小是256 bytes,这时放不下了，只能先发送前128 bytes,等待ack后才能发送剩余部分,这就造成了半包
  3. MSS限制：当发送的数据超过MSS限制后会将数据切分发送，就会造成半包，本质是因为TCP是流式协议，消息无边界

假设客户端分别发送了两个数据包D1和D2给服务端，由于服务端一次读取到字节数是不确定的，故可能存在以下四种情况：
1、服务端分两次读取到了 两个独立的数据包，分别是D1和D2，没有粘包和拆包
2、服务端一次接受到了两个数据包，D1和D2粘合在一起，称之为TCP粘包
3、服务端分两次读取到了数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，这称之为TCP拆包
4、服务端分两次读取到了数据包，第一次读取到了D1包的部分内容D1 1，第二次读取到了D1包的剩余部分内容D1 2和完整的D2，这称之为TCP拆包
image.png
拆包与粘包示意图

粘包拆包演示：

客户端向服务端循环发送数据，模拟出拆包粘包情况，源码地址

客户端：

  1. package tcp;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.ChannelPipeline;
  7. import io.netty.channel.EventLoopGroup;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  12. public class MyTcpClient {
  15.     public static void main(String[] args)  throws  Exception{
  17.         EventLoopGroup group = new NioEventLoopGroup();
  19.         try {
  20.             Bootstrap bootstrap = new Bootstrap();
  21.             bootstrap.group(group).channel(NioSocketChannel.class)
  22.                     .handler(new ChannelInitializer<SocketChannel>() {
  23.                         @Override
  24.                         protected void initChannel(SocketChannel ch) throws Exception {
  25.                             ChannelPipeline pipeline = ch.pipeline();
  26.                             pipeline.addLast(new MyClientHandler());
  27.                         }
  28.                     });
  30.             ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
  31.             channelFuture.channel().closeFuture().sync();
  33.         }finally {
  34.             group.shutdownGracefully();
  35.         }
  37.     }
  40. }

客户端Handler：

  1. package tcp;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.Unpooled;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.SimpleChannelInboundHandler;
  8. import io.netty.util.CharsetUtil;
  10. import java.util.concurrent.atomic.AtomicInteger;
  13. /* 自定义客户端Handler */
  14. public class MyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
  16.     private int sum;
  18.     @Override
  19.     protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  21.         byte[] buffer = new byte[msg.readableBytes()];
  22.         msg.readBytes(buffer);
  23.         String message = new String(buffer, CharsetUtil.UTF_8);
  24.         System.out.println("客户端接收到消息=" +message+" 当前次数:"+ (++this.sum) );
  26.     }
  29.     @Override
  30.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  32.         //循环发送数据模拟粘包
  33.         for (int i = 0; i < 10; ++i) {
  34.             ByteBuf byteBuf = Unpooled.copiedBuffer("hello,server" + i, CharsetUtil.UTF_8);
  35.             System.out.println("客户端发送消息，当前是第"+i+"次");
  36.             ctx.writeAndFlush(byteBuf);
  37.         }
  39.     }
  42.     /* 异常时触发 */
  43.     @Override
  44.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  45.         cause.printStackTrace();
  46.         ctx.close();
  47.     }
  50. }

服务端：

  1. package tcp;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.ChannelPipeline;
  8. import io.netty.channel.EventLoopGroup;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.SocketChannel;
  11. import io.netty.channel.socket.nio.NioServerSocketChannel;
  14. public class MyTcpServer {
  17.     public static void main(String[] args) throws Exception{
  19.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  20.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  22.         try {
  23.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  24.             serverBootstrap.group(bossGroup,workerGroup)
  25.                     .channel(NioServerSocketChannel.class)
  26.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  27.                         @Override
  28.                         protected void initChannel(SocketChannel ch) throws Exception {
  29.                             ChannelPipeline pipeline = ch.pipeline();
  30.                             pipeline.addLast(new MyServerHandler());
  31.                         }
  32.                     });
  34.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  35.             channelFuture.channel().closeFuture().sync();
  36.         }finally {
  37.             bossGroup.shutdownGracefully();
  38.             workerGroup.shutdownGracefully();
  39.         }
  41.     }
  43. }

服务端Handler：

  1. package tcp;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.Unpooled;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.SimpleChannelInboundHandler;
  8. import io.netty.util.CharsetUtil;
  10. import java.util.UUID;
  11. import java.util.concurrent.atomic.AtomicInteger;
  14. /* 自定义服务端Handler */
  15. public class MyServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
  17.     private int sum;
  19.     @Override
  20.     protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  22.         byte[] buffer = new byte[msg.readableBytes()];
  23.         msg.readBytes(buffer);
  25.         String message = new String(buffer, CharsetUtil.UTF_8);  //将buffer转成字符串
  26.         System.out.println("服务器接收到数据: "+ message);
  27.         System.out.println("服务器接收到消息量= " + (++this.sum) );
  29.         //服务器返回数据给客户端
  30.         ByteBuf responseBytebuf = Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8);
  31.         ctx.writeAndFlush(responseBytebuf);
  33.     }
  36. }

测试：

启动客户端与服务端，多次尝试后将出现粘包现象
image.png
粘包场景还原

解决方案：

使用 handler 时只有源码中声明了 @Sharable 才具有线程安全特性

解决方案 - 自定义协议包：

使用自定义协议 + 编解码器解决，核心在于确定服务端每次读取的数据长度，通过将数据封装到协议包来保证每次需要读取的数据内容，源码地址

自定义协议包：

  1. package tcp.protocoltcp;
  3. import lombok.Getter;
  4. import lombok.Setter;
  5. import lombok.experimental.Accessors;
  7. /* 自定义协议包 */
  8. @Getter
  9. @Setter
  10. @Accessors(chain = true)
  11. public class MessageProtocol {
  13.     private int len;    //定义长度
  14.     private byte[] content;     //数据内容
  16. }

自定义编码器：

将需要发送的数据内容封装到协议包内在进行发送

  1. package tcp.protocoltcp;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.MessageToByteEncoder;
  8. /* 自定义协议包对应的编码器 */
  9. public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {
  11.     @Override
  12.     protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
  13.         System.out.println("自定义协议包对应编码器被调用");
  14.         out.writeInt(msg.getLen());
  15.         out.writeBytes(msg.getContent());
  16.     }
  19. }

自定义解码器：

将协议包的数据进行解析

  1. package tcp.protocoltcp;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.handler.codec.ReplayingDecoder;
  7. import java.util.List;
  9. /* 自定义协议包对应的解码器 */
  10. public class MyMessageDecoder extends ReplayingDecoder<Void> {
  12.     @Override
  13.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  14.         System.out.println("自定义协议包对应解码器被调用");
  15.         //需要将得到的二进制字节码转为数据包
  16.         int length = in.readInt();
  17.         byte[] content = new byte[length];
  18.         in.readBytes(content);
  20.         MessageProtocol protocol = new MessageProtocol()
  21.                 .setLen(length)
  22.                 .setContent(content);
  24.         out.add(protocol);
  26.     }
  28. }

服务端Handler：

  1. package tcp.protocoltcp;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.util.CharsetUtil;
  9. /* 自定义客户端Handler */
  10. public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {
  12.     private int sum;
  14.     @Override
  15.     protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
  16.         byte[] content = msg.getContent();
  17.         int len = msg.getLen();
  18.         String message = new String(content, CharsetUtil.UTF_8);  //将buffer转成字符串
  19.         System.out.println("客户端接收到服务器返回数据: "+ message+" 当前长度: "+len);
  20.         System.out.println("客户端接收到消息量= " + (++this.sum) );
  21.     }
  24.     @Override
  25.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  27.         //循环发送数据
  28.         for (int i = 0; i < 5; ++i) {
  29.             String msg = "当前是第" + i+"次签到";
  30.             byte[] bytes = msg.getBytes(CharsetUtil.UTF_8);
  31.             int length = msg.getBytes(CharsetUtil.UTF_8).length;
  33.             //创建协议包对象
  34.             MessageProtocol protocol = new MessageProtocol()
  35.                     .setContent(bytes)
  36.                     .setLen(length);
  37.             System.out.println("客户端发送消息，当前是第"+i+"次");
  38.             ctx.writeAndFlush(protocol);
  39.         }
  41.     }
  44.     /* 异常时触发 */
  45.     @Override
  46.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  47.         cause.printStackTrace();
  48.         ctx.close();
  49.     }
  52. }

客户端Handler：

  1. package tcp.protocoltcp;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.Unpooled;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.SimpleChannelInboundHandler;
  8. import io.netty.util.CharsetUtil;
  10. import java.util.UUID;
  13. /* 自定义服务端Handler */
  14. public class MyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
  16.     private int sum;
  19.     @Override
  20.     protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
  22.         byte[] content = msg.getContent();
  23.         int len = msg.getLen();
  24.         String message = new String(content, CharsetUtil.UTF_8);  //将buffer转成字符串
  25.         System.out.println("服务器接收到数据: "+ message+" 当前长度: "+len);
  26.         System.out.println("服务器接收到消息量= " + (++this.sum) );
  28.         //服务器返回数据给客户端
  29.         String responseStr = UUID.randomUUID().toString();
  30.         int length = responseStr.getBytes("utf-8").length;
  31.         byte[] bytes = responseStr.getBytes();
  32.         MessageProtocol protocol = new MessageProtocol()
  33.                 .setLen(length)
  34.                 .setContent(bytes);
  35.         ctx.writeAndFlush(protocol);
  37.     }
  40. }

客户端：

  1. package tcp.protocoltcp;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.ChannelPipeline;
  7. import io.netty.channel.EventLoopGroup;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  13. /* 模拟出现Tcp粘包情况的客户端 */
  14. public class MyTcpClient {
  17.     public static void main(String[] args)  throws  Exception{
  19.         EventLoopGroup group = new NioEventLoopGroup();
  21.         try {
  22.             Bootstrap bootstrap = new Bootstrap();
  23.             bootstrap.group(group).channel(NioSocketChannel.class)
  24.                     .handler(new ChannelInitializer<SocketChannel>() {
  25.                         @Override
  26.                         protected void initChannel(SocketChannel ch) throws Exception {
  27.                             ChannelPipeline pipeline = ch.pipeline();
  28.                             pipeline.addLast(new MyMessageEncoder());           //自定义协议包对应的编码器
  29.                             pipeline.addLast(new MyMessageDecoder());           //自定义协议包对应的解码器
  30.                             pipeline.addLast(new MyClientHandler());
  31.                         }
  32.                     });
  34.             ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
  35.             channelFuture.channel().closeFuture().sync();
  37.         }finally {
  38.             group.shutdownGracefully();
  39.         }
  41.     }
  44. }

服务端：

  1. package tcp.protocoltcp;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.ChannelPipeline;
  8. import io.netty.channel.EventLoopGroup;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.SocketChannel;
  11. import io.netty.channel.socket.nio.NioServerSocketChannel;
  13. /* 模拟出现Tcp粘包情况的服务端 */
  14. public class MyTcpServer {
  17.     public static void main(String[] args) throws Exception{
  19.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  20.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  22.         try {
  23.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  24.             serverBootstrap.group(bossGroup,workerGroup)
  25.                     .channel(NioServerSocketChannel.class)
  26.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  27.                         @Override
  28.                         protected void initChannel(SocketChannel ch) throws Exception {
  29.                             ChannelPipeline pipeline = ch.pipeline();
  30.                             pipeline.addLast(new MyMessageDecoder());       //自定义协议包的解码器
  31.                             pipeline.addLast(new MyMessageEncoder());       //自定义协议包的编码器
  32.                             pipeline.addLast(new MyServerHandler());
  33.                         }
  34.                     });
  36.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  37.             channelFuture.channel().closeFuture().sync();
  38.         }finally {
  39.             bossGroup.shutdownGracefully();
  40.             workerGroup.shutdownGracefully();
  41.         }
  43.     }
  45. }

测试：

启动服务端和客户端，消息将先通过协议包到达编码器和解码器，每次都只会读取协议包内的数据来实现粘包&拆包问题的解决
image.png
拆包粘包问题得到解决

解决方案 - 定长解码器：

使用定长解码器FixedLengthFrameDecoder对完整数据包的长度进行指定，只有符合长度时才会被获取到，不符合长度则会等待下一次数据包进行数据拼接，有线程安全问题，源码地址
image.png
定长解码器源码中的注释

服务端：

  1. package tcp.fixedlengthframe;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.buffer.ByteBuf;
  6. import io.netty.buffer.Unpooled;
  7. import io.netty.channel.*;
  8. import io.netty.channel.nio.NioEventLoopGroup;
  9. import io.netty.channel.socket.SocketChannel;
  10. import io.netty.channel.socket.nio.NioServerSocketChannel;
  11. import io.netty.handler.codec.FixedLengthFrameDecoder;
  12. import io.netty.util.CharsetUtil;
  14. import java.nio.charset.StandardCharsets;
  15. import java.util.UUID;
  18. /* 使用定长解码器解决拆包粘包问题 - 服务端 */
  19. public class MyTcpServer2 {
  22.     public static void main(String[] args) throws Exception{
  24.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  25.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  27.         try {
  28.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  29.             serverBootstrap.group(bossGroup,workerGroup)
  30.                     .channel(NioServerSocketChannel.class)
  31.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  32.                         @Override
  33.                         protected void initChannel(SocketChannel ch) throws Exception {
  34.                             ChannelPipeline pipeline = ch.pipeline();
  35.                             pipeline.addLast(new FixedLengthFrameDecoder(13));        //添加定长解码器，指定数据长度为10
  36.                             pipeline.addLast(new SimpleChannelInboundHandler<ByteBuf>() {
  38.                                 @Override
  39.                                 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  40.                                     String message = msg.toString(CharsetUtil.UTF_8);
  41.                                     System.out.println("服务器接收到数据: "+ message);
  42.                                 }
  44.                             });
  45.                         }
  46.                     });
  48.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  49.             channelFuture.channel().closeFuture().sync();
  50.         }finally {
  51.             bossGroup.shutdownGracefully();
  52.             workerGroup.shutdownGracefully();
  53.         }
  55.     }
  57. }

客户端：

  1. package tcp.fixedlengthframe;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.Unpooled;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioSocketChannel;
  10. import io.netty.util.CharsetUtil;
  12. import java.nio.charset.StandardCharsets;
  15. /* 使用定长解码器解决拆包粘包问题 - 客户端 */
  16. public class MyTcpClient2 {
  19.     public static void main(String[] args)  throws  Exception{
  21.         EventLoopGroup group = new NioEventLoopGroup();
  23.         try {
  24.             Bootstrap bootstrap = new Bootstrap();
  25.             bootstrap.group(group).channel(NioSocketChannel.class)
  26.                     .handler(new ChannelInitializer<SocketChannel>() {
  27.                         @Override
  28.                         protected void initChannel(SocketChannel ch) throws Exception {
  29.                             ChannelPipeline pipeline = ch.pipeline();
  30.                             pipeline.addLast(new ChannelInboundHandlerAdapter(){
  32.                                 @Override
  33.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  34.                                     //循环发送数据模拟粘包
  35.                                     for (int i = 0; i < 10; ++i) {
  36.                                         ByteBuf buffer = ctx.alloc().buffer();
  37.                                         buffer.writeBytes(("hello,server"+i).getBytes(StandardCharsets.UTF_8));
  38.                                         ctx.writeAndFlush(buffer);
  39.                                     }
  40.                                 }
  42.                             });
  43.                         }
  44.                     });
  46.             ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
  47.             channelFuture.channel().closeFuture().sync();
  49.         }finally {
  50.             group.shutdownGracefully();
  51.         }
  53.     }
  56. }

测试：

定长解码器将数据包进行长度分割，成功获取数据
image.png

解决方案 - 分隔符解码器：

使用分隔符解码器LineBasedFrameDecoder对数据包的长度进行分割，每个数据末尾需要有分割符，实际开发中较为少用，有线程安全问题，源码地址
image.png
分隔符解码器源码使用介绍

服务端：

  1. package tcp.linebaseframe;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.codec.FixedLengthFrameDecoder;
  10. import io.netty.handler.codec.LineBasedFrameDecoder;
  11. import io.netty.util.CharsetUtil;
  14. /* 使用分隔符解码器解决拆包粘包问题 - 服务端 */
  15. public class MyTcpServer3 {
  18.     public static void main(String[] args) throws Exception{
  20.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  21.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  23.         try {
  24.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  25.             serverBootstrap.group(bossGroup,workerGroup)
  26.                     .channel(NioServerSocketChannel.class)
  27.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  28.                         @Override
  29.                         protected void initChannel(SocketChannel ch) throws Exception {
  30.                             ChannelPipeline pipeline = ch.pipeline();
  31.                             pipeline.addLast(new LineBasedFrameDecoder(1024));        //添加分隔符解码器，需要设置数据包最大长度
  32.                             pipeline.addLast(new SimpleChannelInboundHandler<ByteBuf>() {
  34.                                 @Override
  35.                                 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  36.                                     String message = msg.toString(CharsetUtil.UTF_8);
  37.                                     System.out.println("服务器接收到数据: "+ message);
  38.                                 }
  40.                             });
  41.                         }
  42.                     });
  44.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  45.             channelFuture.channel().closeFuture().sync();
  46.         }finally {
  47.             bossGroup.shutdownGracefully();
  48.             workerGroup.shutdownGracefully();
  49.         }
  51.     }
  53. }

客户端：

  1. package tcp.linebaseframe;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  10. import java.nio.charset.StandardCharsets;
  13. /* 使用分隔符解码器解决拆包粘包问题 - 客户端 */
  14. public class MyTcpClient3 {
  17.     public static void main(String[] args)  throws  Exception{
  19.         EventLoopGroup group = new NioEventLoopGroup();
  21.         try {
  22.             Bootstrap bootstrap = new Bootstrap();
  23.             bootstrap.group(group).channel(NioSocketChannel.class)
  24.                     .handler(new ChannelInitializer<SocketChannel>() {
  25.                         @Override
  26.                         protected void initChannel(SocketChannel ch) throws Exception {
  27.                             ChannelPipeline pipeline = ch.pipeline();
  28.                             pipeline.addLast(new ChannelInboundHandlerAdapter(){
  30.                                 @Override
  31.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  32.                                     //循环发送数据，末尾需要使用分隔符，这样服务端分隔符解码器才能正确识别
  33.                                     for (int i = 0; i < 10; ++i) {
  34.                                         ByteBuf buffer = ctx.alloc().buffer();
  35.                                         buffer.writeBytes(("hello,server"+i+"\n").getBytes(StandardCharsets.UTF_8));
  36.                                         ctx.writeAndFlush(buffer);
  37.                                     }
  38.                                 }
  40.                             });
  41.                         }
  42.                     });
  44.             ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
  45.             channelFuture.channel().closeFuture().sync();
  47.         }finally {
  48.             group.shutdownGracefully();
  49.         }
  51.     }
  54. }

测试：

分隔符解码器根据分隔符对数据包进行分割，成功获取数据
image.png

解决方案 - 长度分割解码器：

使用长度分割解码器 LengthFieldBasedFrameDecoder 对数据包进行切割，可以拆出数据包内的指定索引位的数据，有线程安全问题，源码地址
主要参数如下所示：

参数名 含义
maxFrameLength 单个数据包最大长度
lengthFieldOffset 长度字段起始索引位
lengthFieldLength 长度字段长度（16进制长度）
lengthAdjustment 从长度字段结束后开始算，还有几个字节才是内容
initialBytesToStrip 接收数据是从头部开始的几个字节是不需要的

客户端：

  1. package tcp.lengthfieldbasedframe;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.ByteBufAllocator;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioSocketChannel;
  10. import io.netty.util.CharsetUtil;
  12. import java.nio.charset.StandardCharsets;
  14. import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  15. import static io.netty.util.internal.StringUtil.NEWLINE;
  18. /* 使用  解码器解决拆包粘包问题 - 客户端 */
  19. public class MyTcpClient4 {
  22.     public static void main(String[] args)  throws  Exception{
  24.         EventLoopGroup group = new NioEventLoopGroup();
  26.         try {
  27.             Bootstrap bootstrap = new Bootstrap();
  28.             bootstrap.group(group).channel(NioSocketChannel.class)
  29.                     .handler(new ChannelInitializer<SocketChannel>() {
  30.                         @Override
  31.                         protected void initChannel(SocketChannel ch) throws Exception {
  32.                             ChannelPipeline pipeline = ch.pipeline();
  33.                             pipeline.addLast(new ChannelInboundHandlerAdapter(){
  35.                                 @Override
  36.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  37.                                     //循环发送数据
  38.                                     for (int i = 0; i < 10; ++i) {
  39.                                         ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
  40.                                         String info = "hello,server";
  41.                                         byte[] infoBytes = info.getBytes(CharsetUtil.UTF_8);
  42.                                         buffer.writeInt(infoBytes.length);               //写入数据长度，是不需要的信息，将会转换为16进制进行储存，因此服务端截取应该是从4开始
  43.                                         buffer.writeBytes(infoBytes);                    //写入内容
  44.                                         Log(buffer);
  45.                                         ctx.writeAndFlush(buffer);
  46.                                     }
  47.                                 }
  49.                             });
  50.                         }
  51.                     });
  53.             ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
  54.             channelFuture.channel().closeFuture().sync();
  56.         }finally {
  57.             group.shutdownGracefully();
  58.         }
  60.     }
  63.     private static void Log(ByteBuf buffer) {
  64.         int Length = buffer.readableBytes();
  65.         int rows = Length / 16 + (Length % 15 == 0 ? 0 : 1) + 4;
  66.         StringBuilder buf =
  67.                 new StringBuilder(rows * 80 * 2)
  68.                         .append("read index:")
  69.                         .append(buffer.readerIndex())
  70.                         .append(" write index: ")
  71.                         .append(buffer.writerIndex())
  72.                         .append(" capacity:")
  73.                         .append(buffer.capacity())
  74.                         .append(NEWLINE);
  75.         appendPrettyHexDump(buf, buffer);
  76.         System.out.println(buf.toString());
  77.     }
  81. }

服务端：

客户端写入数据转为Byte字节后，前4位为writeInt方法写入的有效数据长度，且接收时数据包的长度位是不需要的，因此 initialBytesToStrip 值为 4，lengthFieldOffset 值为 0，lengthFieldLength 值为 4，长度字段和目标数据字段中间没有其他字段，因此 lengthAdjustment 值为 0 

  1. package tcp.lengthfieldbasedframe;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  9. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
  10. import io.netty.handler.codec.LineBasedFrameDecoder;
  11. import io.netty.handler.codec.string.StringDecoder;
  12. import io.netty.handler.codec.string.StringEncoder;
  13. import io.netty.util.CharsetUtil;
  14. import lombok.extern.java.Log;
  16. import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  17. import static io.netty.util.internal.StringUtil.NEWLINE;
  20. /* 使用  解码器解决拆包粘包问题 - 服务端 */
  21. public class MyTcpServer4 {
  24.     public static void main(String[] args) throws Exception{
  26.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  27.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  29.         try {
  30.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  31.             serverBootstrap.group(bossGroup,workerGroup)
  32.                     .channel(NioServerSocketChannel.class)
  33.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  34.                         @Override
  35.                         protected void initChannel(SocketChannel ch) throws Exception {
  36.                             ChannelPipeline pipeline = ch.pipeline();
  37.                             pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,4,0,4));        //添加分隔符解码器，需要设置数据包最大长度
  38.                             pipeline.addLast(new SimpleChannelInboundHandler<ByteBuf>() {
  40.                                 @Override
  41.                                 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  42.                                     Log(msg);
  43.                                     String message = msg.toString(CharsetUtil.UTF_8);
  44.                                     System.out.println("服务器接收到数据: "+ message);
  45.                                 }
  47.                             });
  48.                         }
  49.                     });
  51.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  52.             channelFuture.channel().closeFuture().sync();
  53.         }finally {
  54.             bossGroup.shutdownGracefully();
  55.             workerGroup.shutdownGracefully();
  56.         }
  58.     }
  61.     private static void Log(ByteBuf buffer) {
  62.         int Length = buffer.readableBytes();
  63.         int rows = Length / 16 + (Length % 15 == 0 ? 0 : 1) + 4;
  64.         StringBuilder buf =
  65.                 new StringBuilder(rows * 80 * 2)
  66.                         .append("read index:")
  67.                         .append(buffer.readerIndex())
  68.                         .append(" write index: ")
  69.                         .append(buffer.writerIndex())
  70.                         .append(" capacity:")
  71.                         .append(buffer.capacity())
  72.                         .append(NEWLINE);
  73.         appendPrettyHexDump(buf, buffer);
  74.         System.out.println(buf.toString());
  75.     }
  79. }

测试：

运行结果可查看具体字节截取效果
image.png