1.分隔符解码器

在日常项目中，我们可能会遇到并不是用回车换行分隔符来修饰包尾，有可能会想自定义分隔符来进行拆包粘包问题的解决，这时就要用到netty里面的DelimiterBasedFrameDecoder解码器来解决上述问题

1.服务端代码

  1. package demo1;
  3. import io.netty.bootstrap.ServerBootstrap;
  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.NioServerSocketChannel;
  10. import io.netty.handler.codec.DelimiterBasedFrameDecoder;
  11. import io.netty.handler.codec.string.StringDecoder;
  13. /**
  14.  * @author 冯铁城 [17615007230@163.com]
  15.  * @date 2022-05-09 19:11:57
  16.  * @describe: 时间服务器服务端
  17.  */
  18. public class TimeServer {
  20.     /**
  21.      * 开启socket链接
  22.      *
  23.      * @param port 端口号
  24.      */
  25.     public void bind(int port) {
  27.         //1.创建线程组用于接受服务端链接
  28.         NioEventLoopGroup bossGroup = new NioEventLoopGroup();
  30.         //2.创建线程组用户socketChannel读写
  31.         NioEventLoopGroup workGroup = new NioEventLoopGroup();
  32.         try {
  34.             //3.创建服务端
  35.             ServerBootstrap server = new ServerBootstrap();
  36.             server.group(bossGroup, workGroup)
  37.                     .channel(NioServerSocketChannel.class)
  38.                     .option(ChannelOption.SO_BACKLOG, 1024)
  39.                     .childHandler(new ChildChannelHandler());
  41.             //3.绑定端口,开启同步等待
  42.             ChannelFuture sync = server.bind(port).sync();
  43.             System.out.println("服务端已启动!");
  45.             //4.等待服务端监听端口关闭
  46.             sync.channel().closeFuture().sync();
  47.         } catch (InterruptedException e) {
  48.             e.printStackTrace();
  49.         } finally {
  50.             bossGroup.shutdownGracefully();
  51.             workGroup.shutdownGracefully();
  52.         }
  53.     }
  55.     private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
  56.         @Override
  57.         protected void initChannel(SocketChannel socketChannel) throws Exception {
  58.             ByteBuf byteBuf = Unpooled.copiedBuffer("-ftc".getBytes());
  59.             socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, byteBuf));
  60.             socketChannel.pipeline().addLast(new StringDecoder());
  61.             socketChannel.pipeline().addLast(new TimeServerHandler());
  62.         }
  63.     }
  65.     private class TimeServerHandler extends ChannelHandlerAdapter {
  67.         @Override
  68.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  69.             System.out.println("服务端已与" + ctx.channel().id() + "建立链接");
  70.         }
  72.         @Override
  73.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  75.             //1.字符串判定
  76.             System.out.println("收到" + ctx.channel().id() + "渠道的请求:" + msg);
  78.             //2.定义返回结果
  79.             String responseMessage = "响应信息-ftc";
  81.             //3.相应数据
  82.             ByteBuf response = Unpooled.copiedBuffer(responseMessage.getBytes());
  83.             ctx.write(response);
  84.         }
  86.         @Override
  87.         public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  88.             ctx.flush();
  89.         }
  91.         @Override
  92.         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  93.             ctx.close();
  94.         }
  95.     }
  96. }

2.客户端代码

  1. package demo1;
  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.handler.codec.DelimiterBasedFrameDecoder;
  11. import io.netty.handler.codec.string.StringDecoder;
  13. /**
  14.  * @author 冯铁城 [17615007230@163.com]
  15.  * @date 2022-05-09 21:09:16
  16.  * @describe: 时间服务器客户端
  17.  */
  18. public class TimeClient {
  20.     /**
  21.      * 创建客户端链接
  22.      *
  23.      * @param host 主机
  24.      * @param port 端口号
  25.      */
  26.     public void connect(String host, int port) {
  28.         //1. 创建用于IO读写的线程组
  29.         EventLoopGroup group = new NioEventLoopGroup();
  30.         try {
  32.             //2.创建客户端启动类
  33.             Bootstrap bootstrap = new Bootstrap();
  35.             //3.启动类初始化
  36.             bootstrap.group(group)
  37.                     .channel(NioSocketChannel.class)
  38.                     .option(ChannelOption.TCP_NODELAY, true)
  39.                     .handler(new ChannelInitializer<SocketChannel>() {
  41.                         @Override
  42.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  43.                             ByteBuf byteBuf = Unpooled.copiedBuffer("-ftc".getBytes());
  44.                             socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, byteBuf));
  45.                             socketChannel.pipeline().addLast(new StringDecoder());
  46.                             socketChannel.pipeline().addLast(new TimeClientHandler());
  47.                         }
  48.                     });
  50.             //4.发起异步链接操作
  51.             ChannelFuture sync = bootstrap.connect(host, port).sync();
  53.             //5.等待客户端链路链接关闭
  54.             sync.channel().closeFuture().sync();
  55.         } catch (InterruptedException e) {
  56.             e.printStackTrace();
  57.         } finally {
  58.             group.shutdownGracefully();
  59.         }
  60.     }
  62.     /**
  63.      * @author 冯铁城 [17615007230@163.com]
  64.      * @date 2022-05-09 21:01:23
  65.      * @describe: 时间服务器客户端
  66.      */
  67.     public class TimeClientHandler extends ChannelHandlerAdapter {
  69.         @Override
  70.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  71.             for (int i = 0; i < 20; i++) {
  72.                 String requestMessage = "发送消息-ftc";
  73.                 ByteBuf message = Unpooled.buffer(requestMessage.getBytes().length);
  74.                 message.writeBytes(requestMessage.getBytes());
  75.                 ctx.writeAndFlush(message);
  76.             }
  77.         }
  79.         @Override
  80.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  81.             System.out.println("收到服务端答复:" + msg);
  82.         }
  84.         @Override
  85.         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  86.             System.out.println("链接出现异常,异常信息:" + cause.getMessage());
  87.             ctx.close();
  88.         }
  89.     }
  90. }

3.调试结果

image.png
image.png
可以看到客户端在发送消息时在包体尾部添加”-ftc”来表示一个数据包的结束，通过DelimiterBasedFrameDecoder实现了拆包粘包问题的解决

4.原理

与LineBasedFrameDecoder原理一样，都是通过遍历字节数组找到对应的包尾标识，来对缓冲区字节数组进行包体划分，进而解决的拆包粘包问题

2.定长解码器

除了在包体增加标识位外，也可以通过消息定长来实现拆包粘包问题的解决。netty中FixedLengthFrameDecoder可以解决上述问题

1.服务端代码

  1. package demo1;
  3. import io.netty.bootstrap.ServerBootstrap;
  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.NioServerSocketChannel;
  10. import io.netty.handler.codec.DelimiterBasedFrameDecoder;
  11. import io.netty.handler.codec.FixedLengthFrameDecoder;
  12. import io.netty.handler.codec.string.StringDecoder;
  14. /**
  15.  * @author 冯铁城 [17615007230@163.com]
  16.  * @date 2022-05-09 19:11:57
  17.  * @describe: 时间服务器服务端
  18.  */
  19. public class TimeServer {
  21.     /**
  22.      * 开启socket链接
  23.      *
  24.      * @param port 端口号
  25.      */
  26.     public void bind(int port) {
  28.         //1.创建线程组用于接受服务端链接
  29.         NioEventLoopGroup bossGroup = new NioEventLoopGroup();
  31.         //2.创建线程组用户socketChannel读写
  32.         NioEventLoopGroup workGroup = new NioEventLoopGroup();
  33.         try {
  35.             //3.创建服务端
  36.             ServerBootstrap server = new ServerBootstrap();
  37.             server.group(bossGroup, workGroup)
  38.                     .channel(NioServerSocketChannel.class)
  39.                     .option(ChannelOption.SO_BACKLOG, 1024)
  40.                     .childHandler(new ChildChannelHandler());
  42.             //3.绑定端口,开启同步等待
  43.             ChannelFuture sync = server.bind(port).sync();
  44.             System.out.println("服务端已启动!");
  46.             //4.等待服务端监听端口关闭
  47.             sync.channel().closeFuture().sync();
  48.         } catch (InterruptedException e) {
  49.             e.printStackTrace();
  50.         } finally {
  51.             bossGroup.shutdownGracefully();
  52.             workGroup.shutdownGracefully();
  53.         }
  54.     }
  56.     private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
  57.         @Override
  58.         protected void initChannel(SocketChannel socketChannel) throws Exception {
  59.             socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(20));
  60.             socketChannel.pipeline().addLast(new StringDecoder());
  61.             socketChannel.pipeline().addLast(new TimeServerHandler());
  62.         }
  63.     }
  65.     private class TimeServerHandler extends ChannelHandlerAdapter {
  67.         @Override
  68.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  69.             System.out.println("服务端已与" + ctx.channel().id() + "建立链接");
  70.         }
  72.         @Override
  73.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  75.             //1.字符串判定
  76.             System.out.println("收到" + ctx.channel().id() + "渠道的请求:" + msg);
  78.             //2.定义返回结果
  79.             String responseMessage = "nihao woshidasahguaa";
  81.             //3.相应数据
  82.             ByteBuf response = Unpooled.copiedBuffer(responseMessage.getBytes());
  83.             ctx.write(response);
  84.         }
  86.         @Override
  87.         public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  88.             ctx.flush();
  89.         }
  91.         @Override
  92.         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  93.             ctx.close();
  94.         }
  95.     }
  96. }

2.客户端代码

  1. package demo1;
  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.handler.codec.DelimiterBasedFrameDecoder;
  11. import io.netty.handler.codec.FixedLengthFrameDecoder;
  12. import io.netty.handler.codec.string.StringDecoder;
  14. /**
  15.  * @author 冯铁城 [17615007230@163.com]
  16.  * @date 2022-05-09 21:09:16
  17.  * @describe: 时间服务器客户端
  18.  */
  19. public class TimeClient {
  21.     /**
  22.      * 创建客户端链接
  23.      *
  24.      * @param host 主机
  25.      * @param port 端口号
  26.      */
  27.     public void connect(String host, int port) {
  29.         //1. 创建用于IO读写的线程组
  30.         EventLoopGroup group = new NioEventLoopGroup();
  31.         try {
  33.             //2.创建客户端启动类
  34.             Bootstrap bootstrap = new Bootstrap();
  36.             //3.启动类初始化
  37.             bootstrap.group(group)
  38.                     .channel(NioSocketChannel.class)
  39.                     .option(ChannelOption.TCP_NODELAY, true)
  40.                     .handler(new ChannelInitializer<SocketChannel>() {
  42.                         @Override
  43.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  44.                             socketChannel.pipeline().addLast(new FixedLengthFrameDecoder(20));
  45.                             socketChannel.pipeline().addLast(new StringDecoder());
  46.                             socketChannel.pipeline().addLast(new TimeClientHandler());
  47.                         }
  48.                     });
  50.             //4.发起异步链接操作
  51.             ChannelFuture sync = bootstrap.connect(host, port).sync();
  53.             //5.等待客户端链路链接关闭
  54.             sync.channel().closeFuture().sync();
  55.         } catch (InterruptedException e) {
  56.             e.printStackTrace();
  57.         } finally {
  58.             group.shutdownGracefully();
  59.         }
  60.     }
  62.     /**
  63.      * @author 冯铁城 [17615007230@163.com]
  64.      * @date 2022-05-09 21:01:23
  65.      * @describe: 时间服务器客户端
  66.      */
  67.     public class TimeClientHandler extends ChannelHandlerAdapter {
  69.         @Override
  70.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  71.             for (int i = 0; i < 20; i++) {
  72.                 String requestMessage = "nihao woshidasahchun";
  73.                 ByteBuf message = Unpooled.buffer(requestMessage.getBytes().length);
  74.                 message.writeBytes(requestMessage.getBytes());
  75.                 ctx.writeAndFlush(message);
  76.             }
  77.         }
  79.         @Override
  80.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  81.             System.out.println("收到服务端答复:" + msg);
  82.         }
  84.         @Override
  85.         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  86.             System.out.println("链接出现异常,异常信息:" + cause.getMessage());
  87.             ctx.close();
  88.         }
  89.     }
  90. }
  91. package demo1;
  93. import io.netty.bootstrap.Bootstrap;
  94. import io.netty.buffer.ByteBuf;
  95. import io.netty.buffer.Unpooled;
  96. import io.netty.channel.*;
  97. import io.netty.channel.nio.NioEventLoopGroup;
  98. import io.netty.channel.socket.SocketChannel;
  99. import io.netty.channel.socket.nio.NioSocketChannel;
  100. import io.netty.handler.codec.DelimiterBasedFrameDecoder;
  101. import io.netty.handler.codec.string.StringDecoder;
  103. /**
  104.  * @author 冯铁城 [17615007230@163.com]
  105.  * @date 2022-05-09 21:09:16
  106.  * @describe: 时间服务器客户端
  107.  */
  108. public class TimeClient {
  110.     /**
  111.      * 创建客户端链接
  112.      *
  113.      * @param host 主机
  114.      * @param port 端口号
  115.      */
  116.     public void connect(String host, int port) {
  118.         //1. 创建用于IO读写的线程组
  119.         EventLoopGroup group = new NioEventLoopGroup();
  120.         try {
  122.             //2.创建客户端启动类
  123.             Bootstrap bootstrap = new Bootstrap();
  125.             //3.启动类初始化
  126.             bootstrap.group(group)
  127.                     .channel(NioSocketChannel.class)
  128.                     .option(ChannelOption.TCP_NODELAY, true)
  129.                     .handler(new ChannelInitializer<SocketChannel>() {
  131.                         @Override
  132.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  133.                             ByteBuf byteBuf = Unpooled.copiedBuffer("-ftc".getBytes());
  134.                             socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, byteBuf));
  135.                             socketChannel.pipeline().addLast(new StringDecoder());
  136.                             socketChannel.pipeline().addLast(new TimeClientHandler());
  137.                         }
  138.                     });
  140.             //4.发起异步链接操作
  141.             ChannelFuture sync = bootstrap.connect(host, port).sync();
  143.             //5.等待客户端链路链接关闭
  144.             sync.channel().closeFuture().sync();
  145.         } catch (InterruptedException e) {
  146.             e.printStackTrace();
  147.         } finally {
  148.             group.shutdownGracefully();
  149.         }
  150.     }
  152.     /**
  153.      * @author 冯铁城 [17615007230@163.com]
  154.      * @date 2022-05-09 21:01:23
  155.      * @describe: 时间服务器客户端
  156.      */
  157.     public class TimeClientHandler extends ChannelHandlerAdapter {
  159.         @Override
  160.         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  161.             for (int i = 0; i < 20; i++) {
  162.                 String requestMessage = "发送消息-ftc";
  163.                 ByteBuf message = Unpooled.buffer(requestMessage.getBytes().length);
  164.                 message.writeBytes(requestMessage.getBytes());
  165.                 ctx.writeAndFlush(message);
  166.             }
  167.         }
  169.         @Override
  170.         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  171.             System.out.println("收到服务端答复:" + msg);
  172.         }
  174.         @Override
  175.         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  176.             System.out.println("链接出现异常,异常信息:" + cause.getMessage());
  177.             ctx.close();
  178.         }
  179.     }
  180. }

3.调试结果

image.png
image.png
如图服务端与客户端分别发送了长度为20的消息，同时设置FixedLengthFrameDecoder的定长为20，客户端与服务端的消息收发完全没有问题

4.原理

通过定长指定消息的长度，只需要按照定长读取缓冲区字节数组即可。读取规定长度后，标志位重置再继续读取下一个包内容