1. 粘包与半包

1.1 粘包现象

客户端一共发送了 10 条消息，每条消息是 10 个字节，而服务端只接收了 1 次

  1. @Slf4j
  2. public class StickyPackageNettyServe {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  6.         NioEventLoopGroup work = new NioEventLoopGroup();
  7.         try {
  8.             ChannelFuture channelFuture = new ServerBootstrap()
  9.                     .group(boos, work)
  10.                     .channel(NioServerSocketChannel.class)
  11.                     .childHandler(new ChannelInitializer<NioSocketChannel>() {
  12.                         @Override
  13.                         protected void initChannel(NioSocketChannel ch) throws Exception {
  14.                             ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  15.                         }
  16.                     })
  17.                     .bind(9527);
  18.             channelFuture.sync();
  19.             channelFuture.channel().closeFuture().sync();
  20.         } catch (Exception e) {
  21.             log.error(e.getMessage(), e);
  22.         } finally {
  23.             boos.shutdownGracefully();
  24.             work.shutdownGracefully();
  25.         }
  26.     }
  28. }
  1. @Slf4j
  2. public class StickyPackageNettyClient {
  4.     public static void main(String[] args) throws Exception {
  5.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  6.         try {
  7.             ChannelFuture channelFuture = new Bootstrap()
  8.                     .group(eventLoopGroup)
  9.                     .channel(NioSocketChannel.class)
  10.                     .handler(new ChannelInitializer<NioSocketChannel>() {
  11.                         @Override
  12.                         protected void initChannel(NioSocketChannel ch) throws Exception {
  13.                             ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  14.                                 @Override
  15.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.                                     // 客户端发送了 10 次数据, 希望服务端也是接收 10 次数据
  17.                                     for (int i = 0; i < 10; i++) {
  18.                                         ByteBuf buf = ctx.alloc().buffer();
  19.                                         buf.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
  20.                                         ctx.writeAndFlush(buf);
  21.                                     }
  22.                                     // 发送完成后关闭通道
  23.                                     ctx.close();
  24.                                 }
  25.                             });
  26.                         }
  27.                     })
  28.                     .connect("127.0.0.1", 9527);
  29.             channelFuture.sync();
  30.             Channel channel = channelFuture.channel();
  31.             channel.closeFuture().sync();
  32.         } catch (InterruptedException e) {
  33.             log.error(e.getMessage(), e);
  34.         } finally {
  35.             eventLoopGroup.shutdownGracefully();
  36.         }
  37.     }
  39. }
  1. 17:41:28.529 [nioEventLoopGroup-3-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x4804e381, L:/127.0.0.1:9527 - R:/127.0.0.1:56867] READ: 100B

1.2 半包现象

客户端发送一条消息，这个消息是 1100 字节，但是服务端接收了两次（1024B + 76B）

  1. @Slf4j
  2. public class HalfPackageNettyServe {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  6.         NioEventLoopGroup work = new NioEventLoopGroup();
  7.         try {
  8.             ChannelFuture channelFuture = new ServerBootstrap()
  9.                 .group(boos, work)
  10.                 .channel(NioServerSocketChannel.class)
  11.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  12.                     @Override
  13.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  14.                         ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  15.                     }
  16.                 })
  17.                 .bind(9527);
  18.             channelFuture.sync();
  19.             channelFuture.channel().closeFuture().sync();
  20.         } catch (Exception e) {
  21.             log.error(e.getMessage(), e);
  22.         } finally {
  23.             boos.shutdownGracefully();
  24.             work.shutdownGracefully();
  25.         }
  26.     }
  28. }
  1. @Slf4j
  2. public class HalfPackageNettyClient {
  4.     public static void main(String[] args) throws Exception {
  5.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  6.         try {
  7.             ChannelFuture channelFuture = new Bootstrap()
  8.                 .group(eventLoopGroup)
  9.                 .channel(NioSocketChannel.class)
  10.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  11.                     @Override
  12.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  13.                         ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  14.                             @Override
  15.                             public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.                                 ByteBuf buffer = ctx.alloc().buffer();
  17.                                 for (int i = 0; i < 110; i++) {
  18.                                     buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
  19.                                 }
  20.                                 // 客户端发送一次数据, 希望服务端也接收一次数据
  21.                                 ctx.writeAndFlush(buffer);
  22.                                 // 发送完成后关闭通道
  23.                                 ctx.close();
  24.                             }
  25.                         });
  26.                     }
  27.                 })
  28.                 .connect("127.0.0.1", 9527);
  29.             channelFuture.sync();
  30.             Channel channel = channelFuture.channel();
  31.             channel.closeFuture().sync();
  32.         } catch (InterruptedException e) {
  33.             log.error(e.getMessage(), e);
  34.         } finally {
  35.             eventLoopGroup.shutdownGracefully();
  36.         }
  37.     }
  39. }
  1. 17:44:05.366 [nioEventLoopGroup-3-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x239806ac, L:/127.0.0.1:9527 - R:/127.0.0.1:57352] READ: 1024B
  2. 17:44:05.369 [nioEventLoopGroup-3-1] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0x239806ac, L:/127.0.0.1:9527 - R:/127.0.0.1:57352] READ: 76B

1.3 现象分析

1.3.1 粘包现象

现象表现是发送 abc def，接收 abcdef

原因分析

  • 应用层：接收方 ByteBuf 设置太大（Netty 默认 1024）
  • 滑动窗口：假设发送方 256 bytes 表示一个完整报文，但由于接收方处理不及时且窗口大小足够大，这 256 bytes 字节就会缓冲在接收方的滑动窗口中，当滑动窗口中缓冲了多个报文就会粘包
  • Nagle 算法：会造成粘包

    1.3.2 半包

    现象表现是发送 abcdef，接收 abc def

原因分析

  • 应用层：接收方 ByteBuf 小于实际发送数据量
  • 滑动窗口：假设接收方的窗口只剩了 128 bytes，发送方的报文大小是 256 bytes，这时放不下了，只能先发送前 128 bytes，等待 ack 后才能发送剩余部分，这就造成了半包

  • MSS 限制：当发送的数据超过 MSS 限制后，会将数据切分发送，就会造成半包

    1.3.3 滑动窗口

    TCP 以一个段（segment）为单位，每发送一个段就需要进行一次确认应答（ack）处理，但如果这么做，缺点是包的往返时间越长性能就越差
    image.png
    为了解决此问题，引入了窗口概念，窗口大小即决定了无需等待应答而可以继续发送的数据最大值
    image.png
    窗口实际就起到一个缓冲区的作用，同时也能起到流量控制的作用

  • 图中深色的部分即要发送的数据，高亮的部分即窗口

  • 窗口内的数据才允许被发送，当应答未到达前，窗口必须停止滑动
  • 如果 1001~2000 这个段的数据 ack 回来了，窗口就可以向前滑动
  • 接收方也会维护一个窗口，只有落在窗口内的数据才能允许接收

    MSS 限制

    • 链路层对一次能够发送的最大数据有限制，这个限制称之为 MTU（maximum transmission unit），不同的链路设备的 MTU 值也有所不同，例如
    • 以太网的 MTU 是 1500
    • FDDI（光纤分布式数据接口）的 MTU 是 4352
    • 本地回环地址的 MTU 是 65535 - 本地测试不走网卡
    • MSS 是最大段长度（maximum segment size），它是 MTU 刨去 tcp 头和 ip 头后剩余能够作为数据传输的字节数
    • ipv4 tcp 头占用 20 bytes，ip 头占用 20 bytes，因此以太网 MSS 的值为 1500 - 40 = 1460
    • TCP 在传递大量数据时，会按照 MSS 大小将数据进行分割发送
    • MSS 的值在三次握手时通知对方自己 MSS 的值，然后在两者之间选择一个小值作为 MSS

Nagle 算法

  • 即使发送一个字节，也需要加入 tcp 头和 ip 头，也就是总字节数会使用 41 bytes，非常不经济。因此为了提高网络利用率，tcp 希望尽可能发送足够大的数据，这就是 Nagle 算法产生的缘由
  • 该算法是指发送端即使还有应该发送的数据，但如果这部分数据很少的话，则进行延迟发送
    • 如果 SO_SNDBUF 的数据达到 MSS，则需要发送
    • 如果 SO_SNDBUF 中含有 FIN（表示需要连接关闭）这时将剩余数据发送，再关闭
    • 如果 TCP_NODELAY = true，则需要发送
    • 已发送的数据都收到 ack 时，则需要发送
    • 上述条件不满足，但发生超时（一般为 200ms）则需要发送
    • 除上述情况，延迟发送

1.4 解决方案

短链接，发一个包建立一次连接，这样连接建立到连接断开之间就是消息的边界，缺点效率太低

每一条消息采用固定长度，缺点浪费空间

每一条消息采用分隔符，例如 \n，缺点需要转义

每一条消息分为 head 和 body，head 中包含 body 的长度

1.4.1 短连接

每次重新创建一个新的连接发送消息

半包用这种办法还是不好解决，因为接收方的缓冲区大小是有限的

  1. public class ShortConnectionNettyServer {
  3.     public static void main(String[] args) {
  4.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  5.         NioEventLoopGroup work = new NioEventLoopGroup();
  6.         try {
  7.             ChannelFuture channelFuture = new ServerBootstrap()
  8.                 .group(boos, work)
  9.                 .channel(NioServerSocketChannel.class)
  10.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  11.                     @Override
  12.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  13.                         ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  14.                     }
  15.                 })
  16.                 .bind(9527);
  17.             channelFuture.sync();
  18.             channelFuture.channel().closeFuture().sync();
  19.         } catch (Exception e) {
  20.             e.printStackTrace();
  21.         } finally {
  22.             boos.shutdownGracefully();
  23.             work.shutdownGracefully();
  24.         }
  25.     }
  27. }
  1. @Slf4j
  2. public class ShortConnectionNettyClient {
  4.     public static void main(String[] args) {
  5.         for (int i = 0; i < 10; i++) {
  6.             sendMessage();
  7.         }
  8.     }
  10.     private static void sendMessage() {
  11.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  12.         try {
  13.             ChannelFuture channelFuture = new Bootstrap()
  14.                 .group(eventLoopGroup)
  15.                 .channel(NioSocketChannel.class)
  16.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  17.                     @Override
  18.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  19.                         ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  20.                             @Override
  21.                             public void channelActive(ChannelHandlerContext ctx) throws Exception {
  22.                                 ByteBuf buffer = ctx.alloc().buffer();
  23.                                 buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
  24.                                 // 客户端发送一次数据, 希望服务端也接收一次数据
  25.                                 ctx.writeAndFlush(buffer);
  26.                                 // 发送完成之后关闭 channel
  27.                                 ctx.close();
  28.                             }
  29.                         });
  30.                     }
  31.                 })
  32.                 .connect("127.0.0.1", 9527);
  33.             channelFuture.sync();
  34.             Channel channel = channelFuture.channel();
  35.             channel.closeFuture().sync();
  36.         } catch (InterruptedException e) {
  37.             log.error(e.getMessage(), e);
  38.         } finally {
  39.             eventLoopGroup.shutdownGracefully();
  40.         }
  41.     }
  43. }

1.4.2 固定长度

让所有数据包长度固定（假设长度为 8 字节），服务器端添加 FixedLengthFrameDecoder ch.pipeline().addLast(new FixedLengthFrameDecoder(8));

缺点是，数据包的大小不好把握

  • 长度定的太大，浪费
  • 长度定的太小，对某些数据包又显得不够
  1. @Slf4j
  2. public class FixedLengthNettyServer {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  6.         NioEventLoopGroup work = new NioEventLoopGroup();
  7.         try {
  8.             ChannelFuture channelFuture = new ServerBootstrap()
  9.                 .group(boos, work)
  10.                 .channel(NioServerSocketChannel.class)
  11.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  12.                     @Override
  13.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  14.                         ch.pipeline().addLast(new FixedLengthFrameDecoder(10));
  15.                         ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  16.                     }
  17.                 })
  18.                 .bind(9527);
  19.             channelFuture.sync();
  20.             channelFuture.channel().closeFuture().sync();
  21.         } catch (Exception e) {
  22.             log.error(e.getMessage(), e);
  23.         } finally {
  24.             boos.shutdownGracefully();
  25.             work.shutdownGracefully();
  26.         }
  27.     }
  29. }
  1. @Slf4j
  2. public class FixedLengthNettyClient {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  6.         try {
  7.             ChannelFuture channelFuture = new Bootstrap()
  8.                 .group(eventLoopGroup)
  9.                 .channel(NioSocketChannel.class)
  10.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  11.                     @Override
  12.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  13.                         ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  14.                             @Override
  15.                             public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.                                 ByteBuf buffer = ctx.alloc().buffer();
  17.                                 for (int i = 0; i < 10; i++) {
  18.                                     buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
  19.                                 }
  20.                                 // 客户端发送一次数据, 希望服务端也接收一次数据
  21.                                 ctx.writeAndFlush(buffer);
  22.                                 // 发送完成之后关闭 channel
  23.                                 ctx.close();
  24.                             }
  25.                         });
  26.                     }
  27.                 })
  28.                 .connect("127.0.0.1", 9527);
  29.             channelFuture.sync();
  30.             Channel channel = channelFuture.channel();
  31.             channel.closeFuture().sync();
  32.         } catch (InterruptedException e) {
  33.             log.error(e.getMessage(), e);
  34.         } finally {
  35.             eventLoopGroup.shutdownGracefully();
  36.         }
  37.     }
  39. }

1.4.3 固定分隔符

服务端加入 LineBasedFrameDecoder，默认以 \n 或 \r\n 作为分隔符，如果超出指定长度仍未出现分隔符，则抛出异常 ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

客户端在每条消息之后，加入 \n 分隔符

  1. @Slf4j
  2. public class LineBasedNettyServer {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  6.         NioEventLoopGroup work = new NioEventLoopGroup();
  7.         try {
  8.             ChannelFuture channelFuture = new ServerBootstrap()
  9.                 .group(boos, work)
  10.                 .channel(NioServerSocketChannel.class)
  11.                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
  12.                     @Override
  13.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  14.                         ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
  15.                         ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  16.                     }
  17.                 })
  18.                 .bind(9527);
  19.             channelFuture.sync();
  20.             channelFuture.channel().closeFuture().sync();
  21.         } catch (Exception e) {
  22.             log.error(e.getMessage(), e);
  23.         } finally {
  24.             boos.shutdownGracefully();
  25.             work.shutdownGracefully();
  26.         }
  27.     }
  29. }
  1. @Slf4j
  2. public class LineBasedNettyClient {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  6.         try {
  7.             ChannelFuture channelFuture = new Bootstrap()
  8.                 .group(eventLoopGroup)
  9.                 .channel(NioSocketChannel.class)
  10.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  11.                     @Override
  12.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  13.                         ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  14.                             @Override
  15.                             public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.                                 ByteBuf buffer = ctx.alloc().buffer();
  17.                                 for (int i = 0; i < 10; i++) {
  18.                                     buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, '\n'});
  19.                                 }
  20.                                 // 客户端发送一次数据, 希望服务端也接收一次数据
  21.                                 ctx.writeAndFlush(buffer);
  22.                                 // 发送完成之后关闭 channel
  23.                                 ctx.close();
  24.                             }
  25.                         });
  26.                     }
  27.                 })
  28.                 .connect("127.0.0.1", 9527);
  29.             channelFuture.sync();
  30.             Channel channel = channelFuture.channel();
  31.             channel.closeFuture().sync();
  32.         } catch (InterruptedException e) {
  33.             log.error(e.getMessage(), e);
  34.         } finally {
  35.             eventLoopGroup.shutdownGracefully();
  36.         }
  37.     }
  39. }

1.4.4 预设长度

在发送消息前，先约定用定长字节表示接下来数据的长度

分别表示最大长度，长度偏移，长度占用字节，长度调整，剥离字节数 ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 1, 0, 1));

  1. @Slf4j
  2. public class LengthFieldBasedNettyServer {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  6.         NioEventLoopGroup work = new NioEventLoopGroup();
  7.         try {
  8.             ChannelFuture channelFuture = new ServerBootstrap()
  9.                     .group(boos, work)
  10.                     .channel(NioServerSocketChannel.class)
  11.                     .childHandler(new ChannelInitializer<NioSocketChannel>() {
  12.                         @Override
  13.                         protected void initChannel(NioSocketChannel ch) throws Exception {
  14.                             ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(2048, 0, 4, 0, 0));
  15.                             ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  16.                         }
  17.                     })
  18.                     .bind(9527);
  19.             channelFuture.sync();
  20.             channelFuture.channel().closeFuture().sync();
  21.         } catch (Exception e) {
  22.             log.error(e.getMessage(), e);
  23.         } finally {
  24.             boos.shutdownGracefully();
  25.             work.shutdownGracefully();
  26.         }
  27.     }
  29. }
  1. @Slf4j
  2. public class LengthFieldBasedNettyClient {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  6.         try {
  7.             ChannelFuture channelFuture = new Bootstrap()
  8.                     .group(eventLoopGroup)
  9.                     .channel(NioSocketChannel.class)
  10.                     .handler(new ChannelInitializer<NioSocketChannel>() {
  11.                         @Override
  12.                         protected void initChannel(NioSocketChannel ch) throws Exception {
  13.                             ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  14.                                 @Override
  15.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.                                     int length = new Random().nextInt(200) + 1000;
  17.                                     log.info("客户端一共准备发送 {} 个字节", length);
  18.                                     char init = 'a';
  19.                                     ByteBuf buffer = ctx.alloc().buffer();
  20.                                     // 先写入 4 个字节的长度
  21.                                     buffer.writeInt(length);
  22.                                     for (int i = 0; i < length; i++) {
  23.                                         if (init == 'z') {
  24.                                             init = 'a';
  25.                                         }
  26.                                         buffer.writeByte(init++);
  27.                                     }
  28.                                     // 客户端发送一次数据, 希望服务端也接收一次数据
  29.                                     ctx.writeAndFlush(buffer);
  30.                                     // 发送完成之后关闭 channel
  31.                                     ctx.close();
  32.                                 }
  33.                             });
  34.                         }
  35.                     })
  36.                     .connect("127.0.0.1", 9527);
  37.             channelFuture.sync();
  38.             Channel channel = channelFuture.channel();
  39.             channel.closeFuture().sync();
  40.         } catch (InterruptedException e) {
  41.             log.error(e.getMessage(), e);
  42.         } finally {
  43.             eventLoopGroup.shutdownGracefully();
  44.         }
  45.     }
  47. }
  1. // 客户端日志
  2. 18:35:31.216 [nioEventLoopGroup-2-1] INFO org.masteryourself.tutorial.netty.lengthfieldbased.LengthFieldBasedNettyClient - 客户端一共准备发送 1193 个字节
  4. // 服务端日志，一共收到 1197B，前 4 个字节表示数据包长度
  5. 18:35:31.251 [nioEventLoopGroup-3-4] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xe699b9a5, L:/127.0.0.1:9527 - R:/127.0.0.1:51315] READ: 1197B

2. 协议设计

什么时候可以加 @Sharable

  • 当 handler 不保存状态时，就可以安全地在多线程下被共享
  • 但要注意对于编解码器类，不能继承 ByteToMessageCodec 或 CombinedChannelDuplexHandler 父类，他们的构造方法对 @Sharable 有限制
  • 如果能确保编解码器不会保存状态，可以继承 MessageToMessageCodec 父类

2.1 redis 协议举例

  1. @Slf4j
  2. public class RedisProtocolNettyClient {
  4.     private static final byte[] LINE = {13, 10};
  6.     public static void main(String[] args) {
  7.         NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
  8.         try {
  9.             ChannelFuture channelFuture = new Bootstrap()
  10.                     .group(eventLoopGroup)
  11.                     .channel(NioSocketChannel.class)
  12.                     .handler(new ChannelInitializer<NioSocketChannel>() {
  13.                         @Override
  14.                         protected void initChannel(NioSocketChannel ch) throws Exception {
  15.                             ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  16.                                 @Override
  17.                                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
  18.                                     set(ctx);
  19.                                     get(ctx);
  20.                                 }
  21.                             });
  22.                         }
  23.                     })
  24.                     .connect("127.0.0.1", 9527);
  25.             channelFuture.sync();
  26.             Channel channel = channelFuture.channel();
  27.             channel.closeFuture().sync();
  28.         } catch (InterruptedException e) {
  29.             log.error(e.getMessage(), e);
  30.         } finally {
  31.             eventLoopGroup.shutdownGracefully();
  32.         }
  33.     }
  35.     private static void get(ChannelHandlerContext ctx) {
  36.         ByteBuf buf = ctx.alloc().buffer();
  37.         buf.writeBytes("*2".getBytes());
  38.         buf.writeBytes(LINE);
  39.         buf.writeBytes("$3".getBytes());
  40.         buf.writeBytes(LINE);
  41.         buf.writeBytes("get".getBytes());
  42.         buf.writeBytes(LINE);
  43.         buf.writeBytes("$3".getBytes());
  44.         buf.writeBytes(LINE);
  45.         buf.writeBytes("aaa".getBytes());
  46.         buf.writeBytes(LINE);
  47.         ctx.writeAndFlush(buf);
  48.     }
  50.     private static void set(ChannelHandlerContext ctx) {
  51.         ByteBuf buf = ctx.alloc().buffer();
  52.         buf.writeBytes("*3".getBytes());
  53.         buf.writeBytes(LINE);
  54.         buf.writeBytes("$3".getBytes());
  55.         buf.writeBytes(LINE);
  56.         buf.writeBytes("set".getBytes());
  57.         buf.writeBytes(LINE);
  58.         buf.writeBytes("$3".getBytes());
  59.         buf.writeBytes(LINE);
  60.         buf.writeBytes("aaa".getBytes());
  61.         buf.writeBytes(LINE);
  62.         buf.writeBytes("$3".getBytes());
  63.         buf.writeBytes(LINE);
  64.         buf.writeBytes("bbb".getBytes());
  65.         buf.writeBytes(LINE);
  66.         ctx.writeAndFlush(buf);
  67.     }
  69. }

2.2 http 协议举例

  1. @Slf4j
  2. public class HttpProtocolNettyServer {
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup boos = new NioEventLoopGroup(1);
  6.         NioEventLoopGroup work = new NioEventLoopGroup();
  7.         try {
  8.             ChannelFuture channelFuture = new ServerBootstrap()
  9.                     .group(boos, work)
  10.                     .channel(NioServerSocketChannel.class)
  11.                     .childHandler(new ChannelInitializer<NioSocketChannel>() {
  12.                         @Override
  13.                         protected void initChannel(NioSocketChannel ch) throws Exception {
  14.                             ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  15.                             ch.pipeline().addLast(new HttpServerCodec());
  16.                             ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
  17.                                 @Override
  18.                                 protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
  19.                                     log.info(msg.uri());
  20.                                     // 返回响应
  21.                                     DefaultFullHttpResponse response =
  22.                                             new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
  23.                                     byte[] bytes = "<h1>you know netty</h1>".getBytes();
  24.                                     response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, bytes.length);
  25.                                     response.content().writeBytes(bytes);
  26.                                     // 写回响应
  27.                                     ctx.writeAndFlush(response);
  28.                                 }
  29.                             });
  30.                         }
  31.                     })
  32.                     .bind(9527);
  33.             channelFuture.sync();
  34.             channelFuture.channel().closeFuture().sync();
  35.         } catch (Exception e) {
  36.             log.error(e.getMessage(), e);
  37.         } finally {
  38.             boos.shutdownGracefully();
  39.             work.shutdownGracefully();
  40.         }
  41.     }
  43. }

2.3 自定义协议

魔数，用来在第一时间判定是否是无效数据包 版本号，可以支持协议的升级 序列化算法，消息正文到底采用哪种序列化反序列化方式，可以由此扩展，例如：json、protobuf、hessian、jdk 指令类型，是登录、注册、单聊、群聊… 跟业务相关 请求序号，为了双工通信，提供异步能力 正文长度 消息正文

  1. @Data
  2. @Builder
  3. public class Message implements Serializable {
  5.     private Byte messageType;
  7.     private Integer sequenceId;
  9.     private String username;
  11.     private String password;
  13. }
  1. @Slf4j
  2. public class MessageCodec extends ByteToMessageCodec<Message> {
  4.     @Override
  5.     protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
  6.         // 1. 4 个字节的魔数
  7.         out.writeBytes(new byte[]{1, 2, 3, 4});
  8.         // 2. 1 个字节的版本
  9.         out.writeByte(1);
  10.         // 3. 1 个字节的序列化方式(jdk 为 0, json 为 1)
  11.         out.writeByte(0);
  12.         // 4. 1 个字节的消息类型
  13.         out.writeByte(msg.getMessageType());
  14.         // 5. 4 个字节的请求序列号
  15.         out.writeInt(msg.getSequenceId());
  16.         // 6. 无意义, 对其填充(共 16 个字节)
  17.         out.writeByte(0xff);
  18.         // 获取内容长度
  19.         ByteArrayOutputStream bos = new ByteArrayOutputStream();
  20.         ObjectOutputStream oos = new ObjectOutputStream(bos);
  21.         oos.writeObject(msg);
  22.         byte[] content = bos.toByteArray();
  23.         // 7. 4 个字节长度
  24.         out.writeInt(content.length);
  25.         // 8. 消息正文
  26.         out.writeBytes(content);
  27.     }
  29.     @Override
  30.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  31.         // 1. 获取魔数
  32.         int magicNum = in.readInt();
  33.         // 2. 获取版本
  34.         byte version = in.readByte();
  35.         // 3. 获取序列化类型
  36.         byte serializerType = in.readByte();
  37.         // 4. 获取消息类型
  38.         byte messageType = in.readByte();
  39.         // 5. 获取请求序列号
  40.         int sequenceId = in.readInt();
  41.         // 6. 无意义字节
  42.         in.readByte();
  43.         // 7. 获取内容长度
  44.         int length = in.readInt();
  45.        // 8. 解析消息正文
  46.         byte[] bytes = new byte[length];
  47.         in.readBytes(bytes, 0, length);
  48.         ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
  49.         Message message = (Message) ois.readObject();
  50.         log.debug("magicNum:{}, version:{}, serializerType:{}, messageType:{}, sequenceId:{}, length:{}",
  51.                 magicNum, version, serializerType, messageType, sequenceId, length);
  52.         log.debug("{}", message);
  53.         out.add(message);
  54.     }
  56. }
  1. public class CustomizeProtocolTest {
  3.     public static void main(String[] args) throws Exception {
  4.         EmbeddedChannel channel = new EmbeddedChannel(
  5.                 new LoggingHandler(),
  6.                 new LengthFieldBasedFrameDecoder(1024, 12, 4, 0, 0),
  7.                 new MessageCodec()
  8.         );
  9.         Message message = Message.builder()
  10.                 .messageType((byte) 1)
  11.                 .sequenceId(10001)
  12.                 .username("zhangsan")
  13.                 .password("123456")
  14.                 .build();
  15.         ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
  16.         new MessageCodec().encode(null, message, buf);
  17.         channel.writeInbound(buf);
  18.     }
  20. }

3. 聊天室案例

3.1 登录

  1. @ChannelHandler.Sharable
  2. public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<org.masteryourself.tutorial.netty.chat.message.LoginRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, org.masteryourself.tutorial.netty.chat.message.LoginRequestMessage msg) throws Exception {
  6.         String userName = msg.getUsername();
  7.         String password = msg.getPassword();
  8.         boolean login = UserServiceFactory.getUserService().login(userName, password);
  9.         if (login) {
  10.             ctx.writeAndFlush(new LoginResponseMessage(true, "登录成功"));
  11.             return;
  12.         }
  13.         ctx.writeAndFlush(new LoginResponseMessage(false, "登录失败"));
  14.     }
  16. }
  1. public class ClientRequestHandler extends ChannelInboundHandlerAdapter {
  3.     private final CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
  4.     private final AtomicBoolean LOGIN_FLAG = new AtomicBoolean(false);
  6.     @Override
  7.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  8.         new Thread(new Runnable() {
  9.             @Override
  10.             public void run() {
  11.                 Scanner scanner = new Scanner(System.in);
  12.                 System.out.println("请输入用户名:");
  13.                 String username = scanner.nextLine();
  14.                 System.out.println("请输入密码:");
  15.                 String password = scanner.nextLine();
  16.                 // 构造消息对象
  17.                 LoginRequestMessage message = new LoginRequestMessage(username, password);
  18.                 // 发送消息
  19.                 ctx.writeAndFlush(message);
  20.                 System.out.println("等待后续操作...");
  21.                 try {
  22.                     WAIT_FOR_LOGIN.await();
  23.                 } catch (InterruptedException e) {
  24.                     e.printStackTrace();
  25.                 }
  26.                 // 如果登录失败
  27.                 if (!LOGIN_FLAG.get()) {
  28.                     System.out.println("登录失败, 系统退出");
  29.                     ctx.channel().close();
  30.                     return;
  31.                 }
  32.                 // 处理业务逻辑
  33.                 while (true) {
  34.                     System.out.println("==================================");
  35.                     System.out.println("send [username] [content]");
  36.                     System.out.println("gsend [group name] [content]");
  37.                     System.out.println("gcreate [group name] [m1,m2,m3...]");
  38.                     System.out.println("gmembers [group name]");
  39.                     System.out.println("gjoin [group name]");
  40.                     System.out.println("gquit [group name]");
  41.                     System.out.println("quit");
  42.                     System.out.println("==================================");
  43.                     String command = scanner.nextLine();
  44.                     String[] s = command.split(" ");
  45.                     switch (s[0]) {
  46.                         case "send":
  47.                             ctx.writeAndFlush(new ChatRequestMessage(username, s[1], s[2]));
  48.                             break;
  49.                         case "gsend":
  50.                             ctx.writeAndFlush(new GroupChatRequestMessage(username, s[1], s[2]));
  51.                             break;
  52.                         case "gcreate":
  53.                             Set<String> set = new HashSet<>(Arrays.asList(s[2].split(",")));
  54.                             set.add(username); // 加入自己
  55.                             ctx.writeAndFlush(new GroupCreateRequestMessage(s[1], set));
  56.                             break;
  57.                         case "gmembers":
  58.                             ctx.writeAndFlush(new GroupMembersRequestMessage(s[1]));
  59.                             break;
  60.                         case "gjoin":
  61.                             ctx.writeAndFlush(new GroupJoinRequestMessage(username, s[1]));
  62.                             break;
  63.                         case "gquit":
  64.                             ctx.writeAndFlush(new GroupQuitRequestMessage(username, s[1]));
  65.                             break;
  66.                         case "quit":
  67.                             ctx.channel().close();
  68.                             return;
  69.                     }
  70.                 }
  71.             }
  72.         }, "SYSTEM IN").start();
  73.         // 继续调用 chain
  74.         super.channelActive(ctx);
  75.     }
  77.     @Override
  78.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  79.         if (msg instanceof LoginResponseMessage) {
  80.             LoginResponseMessage response = (LoginResponseMessage) msg;
  81.             if (response.isSuccess()) {
  82.                 LOGIN_FLAG.set(true);
  83.             }
  84.             WAIT_FOR_LOGIN.countDown();
  85.         }
  86.     }
  88. }

3.2 单聊

  1. @ChannelHandler.Sharable
  2. public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<org.masteryourself.tutorial.netty.chat.message.LoginRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, org.masteryourself.tutorial.netty.chat.message.LoginRequestMessage msg) throws Exception {
  6.         String userName = msg.getUsername();
  7.         String password = msg.getPassword();
  8.         boolean login = UserServiceFactory.getUserService().login(userName, password);
  9.         if (login) {
  10.             // 存储到会话中
  11.             SessionFactory.getSession().bind(ctx.channel(), userName);
  12.             ctx.writeAndFlush(new LoginResponseMessage(true, "登录成功"));
  13.             return;
  14.         }
  15.         ctx.writeAndFlush(new LoginResponseMessage(false, "登录失败"));
  16.     }
  18. }
  1. @ChannelHandler.Sharable
  2. public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {
  6.         String to = msg.getTo();
  7.         Channel channel = SessionFactory.getSession().getChannel(to);
  8.         if (channel != null) {
  9.             // 在线就发送给对方
  10.             channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));
  11.             return;
  12.         }
  13.         // 不在线就提醒给发送人
  14.         ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不在线"));
  15.     }
  17. }

3.3 群聊

  1. @ChannelHandler.Sharable
  2. public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupCreateRequestMessage msg) throws Exception {
  6.         String groupName = msg.getGroupName();
  7.         Set<String> members = msg.getMembers();
  8.         // 先判断名称是否重复
  9.         Group group = GroupSessionFactory.getGroupSession().createGroup(groupName, members);
  10.         if (group != null) {
  11.             ctx.writeAndFlush(new GroupCreateResponseMessage(false, groupName + "已经存在"));
  12.             return;
  13.         }
  14.         // 消息提醒
  15.         for (String member : members) {
  16.             Channel channel = SessionFactory.getSession().getChannel(member);
  17.             if (channel != null) {
  18.                 channel.writeAndFlush(new GroupCreateResponseMessage(true, "您已被拉入" + groupName));
  19.             }
  20.         }
  21.     }
  23. }
  1. @ChannelHandler.Sharable
  2. public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {
  6.         Set<String> members = GroupSessionFactory.getGroupSession().getMembers(msg.getGroupName());
  7.         for (String member : members) {
  8.             Channel channel = SessionFactory.getSession().getChannel(member);
  9.             if (channel != null) {
  10.                 channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));
  11.             }
  12.         }
  13.     }
  15. }
  1. @ChannelHandler.Sharable
  2. public class GroupJoinRequestMessageHandler extends SimpleChannelInboundHandler<GroupJoinRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupJoinRequestMessage msg) throws Exception {
  6.         Group group = GroupSessionFactory.getGroupSession().joinMember(msg.getGroupName(), msg.getUsername());
  7.         if (group == null) {
  8.             ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
  9.             return;
  10.         }
  11.         ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群加入成功"));
  12.     }
  14. }
  1. @ChannelHandler.Sharable
  2. public class GroupQuitRequestMessageHandler extends SimpleChannelInboundHandler<GroupQuitRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupQuitRequestMessage msg) throws Exception {
  6.         Group group = GroupSessionFactory.getGroupSession().removeMember(msg.getGroupName(), msg.getUsername());
  7.         if (group == null) {
  8.             ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
  9.             return;
  10.         }
  11.         ctx.writeAndFlush(new GroupJoinResponseMessage(true, "已退出群" + msg.getGroupName()));
  12.         // 通知其它人
  13.         Set<String> members = GroupSessionFactory.getGroupSession().getMembers(msg.getGroupName());
  14.         for (String member : members) {
  15.             Channel channel = SessionFactory.getSession().getChannel(member);
  16.             if (channel != null) {
  17.                 channel.writeAndFlush(new GroupJoinResponseMessage(true, "已退出群" + msg.getGroupName()));
  18.             }
  19.         }
  20.     }
  22. }
  1. @ChannelHandler.Sharable
  2. public class GroupMembersRequestMessageHandler extends SimpleChannelInboundHandler<GroupMembersRequestMessage> {
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, GroupMembersRequestMessage msg) throws Exception {
  6.         Set<String> members = GroupSessionFactory.getGroupSession()
  7.                 .getMembers(msg.getGroupName());
  8.         ctx.writeAndFlush(new GroupMembersResponseMessage(members));
  9.     }
  11. }

3.4 退出

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class QuitHandler extends ChannelInboundHandlerAdapter {
  5.     /**
  6.      * 连接断开时触发
  7.      */
  8.     @Override
  9.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  10.         SessionFactory.getSession().unbind(ctx.channel());
  11.     }
  13.     /**
  14.      * 出现异常时触发
  15.      */
  16.     @Override
  17.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  18.         SessionFactory.getSession().unbind(ctx.channel());
  19.         log.error(cause.getMessage(), cause);
  20.     }
  22. }

3.5 空闲检测

3.5.1 连接假死

问题

  • 假死的连接占用的资源不能自动释放
  • 向假死的连接发送数据，得到的反馈是发送超时

原因

  • 网络设备出现故障，例如网卡，机房等，底层的 TCP 连接已经断开了，但应用程序没有感知到，仍然占用着资源。
  • 公网网络不稳定，出现丢包。如果连续出现丢包，这时现象就是客户端数据发不出去，服务端也一直收不到数据，就这么一直耗着
  • 应用程序线程阻塞，无法进行数据读写

    3.5.2 服务端解决

    怎么判断客户端连接是否假死呢？如果能收到客户端数据，说明没有假死。因此策略就可以定为，每隔一段时间就检查这段时间内是否接收到客户端数据，没有就可以判定为连接假死

  1. // 用来判断是不是 [读空闲时间过长]，或 [写空闲时间过长]
  2. // 5s 内如果没有收到 channel 的数据，会触发一个 IdleState#READER_IDLE 事件
  3. ch.pipeline().addLast(new IdleStateHandler(5, 0, 0));
  4. // ChannelDuplexHandler 可以同时作为入站和出站处理器
  5. ch.pipeline().addLast(new ChannelDuplexHandler() {
  6.     // 用来触发特殊事件
  7.     @Override
  8.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  9.         IdleStateEvent event = (IdleStateEvent) evt;
  10.         // 触发了读空闲事件
  11.         if (IdleState.READER_IDLE == event.state()) {
  12.             log.info("已经 5s 没有读取到数据了, 干掉连接");
  13.             ctx.channel().close();
  14.         }
  15.     }
  16. });

3.5.3 客户端解决

客户端可以定时向服务器端发送数据，只要这个时间间隔小于服务器定义的空闲检测的时间间隔，那么就能防止前面提到的误判，客户端可以定义如下心跳处理器

  1. // 用来判断是不是[读空闲时间过长]，或[写空闲时间过长]
  2. // 3s 内如果没有向服务器写数据，会触发一个 IdleState#WRITER_IDLE 事件
  3. ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
  4. // ChannelDuplexHandler 可以同时作为入站和出站处理器
  5. ch.pipeline().addLast(new ChannelDuplexHandler() {
  6.     // 用来触发特殊事件
  7.     @Override
  8.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  9.         IdleStateEvent event = (IdleStateEvent) evt;
  10.         // 触发了写空闲事件
  11.         if (event.state() == IdleState.WRITER_IDLE) {
  12.             log.info("3s 没有写数据了，发送一个心跳包");
  13.             ctx.writeAndFlush(new PingMessage());
  14.         }
  15.     }
  16. });