一、黏包与半包

1. 黏包现象演示

服务端代码
  1. public class HelloWorldServer {
  2. static final Logger log = LoggerFactory.getLogger(HelloWorldServer.class);
  3. void start() {
  4. NioEventLoopGroup boss = new NioEventLoopGroup(1);
  5. NioEventLoopGroup worker = new NioEventLoopGroup();
  6. try {
  7. ServerBootstrap serverBootstrap = new ServerBootstrap();
  8. serverBootstrap.channel(NioServerSocketChannel.class);
  9. serverBootstrap.group(boss, worker);
  10. serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  11. @Override
  12. protected void initChannel(SocketChannel ch) throws Exception {
  13. ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  14. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  15. @Override
  16. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  17. log.debug("connected {}", ctx.channel());
  18. super.channelActive(ctx);
  19. }
  20. @Override
  21. public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  22. log.debug("disconnect {}", ctx.channel());
  23. super.channelInactive(ctx);
  24. }
  25. });
  26. }
  27. });
  28. ChannelFuture channelFuture = serverBootstrap.bind(8080);
  29. log.debug("{} binding...", channelFuture.channel());
  30. channelFuture.sync();
  31. log.debug("{} bound...", channelFuture.channel());
  32. channelFuture.channel().closeFuture().sync();
  33. } catch (InterruptedException e) {
  34. log.error("server error", e);
  35. } finally {
  36. boss.shutdownGracefully();
  37. worker.shutdownGracefully();
  38. log.debug("stoped");
  39. }
  40. }
  41. public static void main(String[] args) {
  42. new HelloWorldServer().start();
  43. }
  44. }

客户端代码希望发送 10 个消息,每个消息是 16 字节
  1. public class HelloWorldClient {
  2. static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
  3. public static void main(String[] args) {
  4. NioEventLoopGroup worker = new NioEventLoopGroup();
  5. try {
  6. Bootstrap bootstrap = new Bootstrap();
  7. bootstrap.channel(NioSocketChannel.class);
  8. bootstrap.group(worker);
  9. bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  10. @Override
  11. protected void initChannel(SocketChannel ch) throws Exception {
  12. log.debug("connetted...");
  13. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  14. @Override
  15. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16. log.debug("sending...");
  17. Random r = new Random();
  18. char c = 'a';
  19. for (int i = 0; i < 10; i++) {
  20. ByteBuf buffer = ctx.alloc().buffer();
  21. buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
  22. ctx.writeAndFlush(buffer);
  23. }
  24. }
  25. });
  26. }
  27. });
  28. ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
  29. channelFuture.channel().closeFuture().sync();
  30. } catch (InterruptedException e) {
  31. log.error("client error", e);
  32. } finally {
  33. worker.shutdownGracefully();
  34. }
  35. }
  36. }

打印结果
  1. 08:24:46 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x81e0fda5] binding...
  2. 08:24:46 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x81e0fda5, L:/0:0:0:0:0:0:0:0:8080] bound...
  3. 08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] REGISTERED
  4. 08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] ACTIVE
  5. 08:24:55 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177]
  6. 08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] READ: 160B
  7. +-------------------------------------------------+
  8. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  9. +--------+-------------------------------------------------+----------------+
  10. |00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  11. |00000010| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  12. |00000020| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  13. |00000030| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  14. |00000040| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  15. |00000050| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  16. |00000060| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  17. |00000070| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  18. |00000080| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  19. |00000090| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
  20. +--------+-------------------------------------------------+----------------+
  21. 08:24:55 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x94132411, L:/127.0.0.1:8080 - R:/127.0.0.1:58177] READ COMPLETE

2. 半包现象演示

客户端代码希望发送 1 个消息,这个消息是 160 字节,代码改为

ByteBuf buffer = ctx.alloc().buffer();
for (int i = 0; i < 10; i++) {
    buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
}
ctx.writeAndFlush(buffer);

为现象明显,服务端修改一下接收缓冲区,其它代码不变

serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);

服务器端的某次输出,可以看到接收的消息被分为两节,第一次 20 字节,第二次 140 字节

08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84] binding...
08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84, L:/0:0:0:0:0:0:0:0:8080] bound...
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] REGISTERED
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] ACTIVE
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221]
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ: 20B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000010| 00 01 02 03                                     |....            |
+--------+-------------------------------------------------+----------------+
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ COMPLETE
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ: 140B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000010| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000020| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000030| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000040| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000050| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000060| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000070| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000080| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f             |............    |
+--------+-------------------------------------------------+----------------+
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ COMPLETE

补充

serverBootstrap.option(ChannelOption.SO_RCVBUF, 10) 影响的底层接收缓冲区(即滑动窗口)大小,仅决定了 netty 读取的最小单位,netty 实际每次读取的一般是它的整数倍

3. 现象分析

1. 黏包

现象:发送 abc def,接收 abcdef
原因:

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

2. 半包

现象:发送 abcdef,接收 abc def
原因

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

本质是因为 TCP 是流式协议,消息无边界

3. 滑动窗口

TCP 以一个段(segment)为单位,每发送一个段就需要进行一次确认应答(ack)处理,但如果这么做,缺点是包的往返时间越长性能就越差
0049.png 0028.jpg

为了解决此问题,引入了窗口概念,窗口大小即决定了无需等待应答而可以继续发送的数据最大值
0051.png 0029.jpg
0050.png 0030.jpg

窗口实际就起到一个缓冲区的作用,同时也能起到流量控制的作用

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

4. 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

0031.jpg

5. Nagle算法

  • 即使发送一个字节,也需要加入 tcp 头和 ip 头,也就是总字节数会使用 41 bytes,非常不经济。因此为了提高网络利用率,tcp 希望尽可能发送足够大的数据,这就是 Nagle 算法产生的缘由

  • 该算法是指发送端即使还有应该发送的数据,但如果这部分数据很少的话,则进行延迟发送

    • 如果 SO_SNDBUF 的数据达到 MSS,则需要发送
    • 如果 SO_SNDBUF 中含有 FIN(表示需要连接关闭)这时将剩余数据发送,再关闭
    • 如果 TCP_NODELAY = true,则需要发送
    • 已发送的数据都收到 ack 时,则需要发送
    • 上述条件不满足,但发生超时(一般为 200ms)则需要发送
    • 除上述情况,延迟发送

6. 解决方案

  1. 短链接,发一个包建立一次连接,这样连接建立到连接断开之间就是消息的边界,缺点效率太低
  2. 每一条消息采用固定长度,缺点浪费空间
  3. 每一条消息采用分隔符,例如 \n,缺点需要转义
  4. 每一条消息分为 head 和 body,head 中包含 body 的长度

方案一:短链接

以解决粘包为例

public class HelloWorldClient {
    static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);

    public static void main(String[] args) {
        // 分 10 次发送
        for (int i = 0; i < 10; i++) {
            send();
        }
    }

    private static void send() {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    log.debug("conneted...");
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            ByteBuf buffer = ctx.alloc().buffer();
                            buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
                            ctx.writeAndFlush(buffer);
                            // 发完即关
                            ctx.close();
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

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

方案二:固定长度

服务器端加入下面处理器,让所有数据包长度固定(假设长度为 8 字节)

ch.pipeline().addLast(new FixedLengthFrameDecoder(8));

客户端测试代码,注意, 采用这种方法后,客户端什么时候 flush 都可以

public class HelloWorldClient {
    static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);

    public static void main(String[] args) {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    log.debug("connetted...");
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            // 发送内容随机的数据包
                            Random r = new Random();
                            char c = 'a';
                            ByteBuf buffer = ctx.alloc().buffer();
                            for (int i = 0; i < 10; i++) {
                                byte[] bytes = new byte[8];
                                for (int j = 0; j < r.nextInt(8); j++) {
                                    bytes[j] = (byte) c;
                                }
                                c++;
                                buffer.writeBytes(bytes);
                            }
                            ctx.writeAndFlush(buffer);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

客户端输出

12:07:00 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2] REGISTERED
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2] CONNECT: /localhost:8080
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2, L:/localhost:53155 - R:/localhost:8080] ACTIVE
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2, L:/localhost:53155 - R:/localhost:8080] WRITE: 80B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 00 00 00 00 62 00 00 00 00 00 00 00 |aaaa....b.......|
|00000010| 63 63 00 00 00 00 00 00 64 00 00 00 00 00 00 00 |cc......d.......|
|00000020| 00 00 00 00 00 00 00 00 66 66 66 66 00 00 00 00 |........ffff....|
|00000030| 67 67 67 00 00 00 00 00 68 00 00 00 00 00 00 00 |ggg.....h.......|
|00000040| 69 69 69 69 69 00 00 00 6a 6a 6a 6a 00 00 00 00 |iiiii...jjjj....|
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x3c2ef3c2, L:/localhost:53155 - R:/localhost:8080] FLUSH

服务端输出

12:06:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xe3d9713f] binding...
12:06:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xe3d9713f, L:/192.168.0.103:9090] bound...
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] REGISTERED
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] ACTIVE
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155]
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 61 61 61 61 00 00 00 00                         |aaaa....        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 62 00 00 00 00 00 00 00                         |b.......        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 63 63 00 00 00 00 00 00                         |cc......        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 64 00 00 00 00 00 00 00                         |d.......        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 00 00 00 00 00                         |........        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 66 66 66 00 00 00 00                         |ffff....        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 67 67 67 00 00 00 00 00                         |ggg.....        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 00 00 00 00 00 00 00                         |h.......        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 69 69 69 69 69 00 00 00                         |iiiii...        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ: 8B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 6a 6a 6a 6a 00 00 00 00                         |jjjj....        |
+--------+-------------------------------------------------+----------------+
12:07:00 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xd739f137, L:/192.168.0.103:9090 - R:/192.168.0.103:53155] READ COMPLETE

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

  • 长度定的太大,浪费
  • 长度定的太小,对某些数据包又显得不够

方案三:固定分隔符

服务端加入,默认以 \n 或 \r\n 作为分隔符,如果超出指定长度仍未出现分隔符,则抛出异常
image.png

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

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

@Slf4j
public class nianBaoBanBaoClient {
    public static void main(String[] args) {
        send();
        System.out.println("finish");
    }

    public static StringBuilder makeString(char c, int len) {
        StringBuilder sb = new StringBuilder(len + 2);
        for (int i = 0; i < len; i++) {
            sb.append(c);
        }
        sb.append("\n");
        return sb;
    }

    private static void send() {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {

                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        // 会在连接 channel 建立成功后,会触发 active 事件
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) {
                            ByteBuf buf = ctx.alloc().buffer();
                            char c = '0';
                            Random r = new Random();
                            for (int i = 0; i < 10; i++) {
                                StringBuilder sb = makeString(c, r.nextInt(256) + 1);
                                c++;
                                buf.writeBytes(sb.toString().getBytes());
                            }
                            ctx.writeAndFlush(buf);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

缺点,处理字符数据比较合适,但如果内容本身包含了分隔符(字节数据常常会有此情况),那么就会解析错误

方案四:预设长度

io.netty.handler.codec.LengthFieldBasedFrameDecoder 官方案例

image.png
image.png
image.png

参数解析
  • lengthFieldoffset:长度字段偏移量
  • lengthFieldLength:长度字段长度
  • lengthAdjustment:长度字段为基准,还有几个字节是内容
  • initialBytesTostrip:从头剥离几个字节

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

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

客户端代码

public class HelloWorldClient {
    static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);

    public static void main(String[] args) {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    log.debug("connetted...");
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            Random r = new Random();
                            char c = 'a';
                            ByteBuf buffer = ctx.alloc().buffer();
                            for (int i = 0; i < 10; i++) {
                                byte length = (byte) (r.nextInt(16) + 1);
                                // 先写入长度
                                buffer.writeByte(length);
                                // 再
                                for (int j = 1; j <= length; j++) {
                                    buffer.writeByte((byte) c);
                                }
                                c++;
                            }
                            ctx.writeAndFlush(buffer);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

客户端输出

 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8] REGISTERED
 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8] CONNECT: /192.168.0.103:9090
 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8, L:/192.168.0.103:49979 - R:/192.168.0.103:9090] ACTIVE
 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8, L:/192.168.0.103:49979 - R:/192.168.0.103:9090] WRITE: 97B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 09 61 61 61 61 61 61 61 61 61 09 62 62 62 62 62 |.aaaaaaaaa.bbbbb|
 |00000010| 62 62 62 62 06 63 63 63 63 63 63 08 64 64 64 64 |bbbb.cccccc.dddd|
 |00000020| 64 64 64 64 0f 65 65 65 65 65 65 65 65 65 65 65 |dddd.eeeeeeeeeee|
 |00000030| 65 65 65 65 0d 66 66 66 66 66 66 66 66 66 66 66 |eeee.fffffffffff|
 |00000040| 66 66 02 67 67 02 68 68 0e 69 69 69 69 69 69 69 |ff.gg.hh.iiiiiii|
 |00000050| 69 69 69 69 69 69 69 09 6a 6a 6a 6a 6a 6a 6a 6a |iiiiiii.jjjjjjjj|
 |00000060| 6a                                              |j               |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8, L:/192.168.0.103:49979 - R:/192.168.0.103:9090] FLUSH

服务端输出

 14:36:50 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xdff439d3] binding...
 14:36:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xdff439d3, L:/192.168.0.103:9090] bound...
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] REGISTERED
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] ACTIVE
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979]
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 9B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 61 61 61 61 61 61 61 61 61                      |aaaaaaaaa       |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 9B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 62 62 62 62 62 62 62 62 62                      |bbbbbbbbb       |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 6B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 63 63 63 63 63 63                               |cccccc          |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 8B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 64 64 64 64 64 64 64 64                         |dddddddd        |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 15B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 65 65 65 65 65 65 65 65 65 65 65 65 65 65 65    |eeeeeeeeeeeeeee |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 13B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 66 66 66 66 66 66 66 66 66 66 66 66 66          |fffffffffffff   |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 2B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 67 67                                           |gg              |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 2B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 68 68                                           |hh              |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 14B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 69 69 69 69 69 69 69 69 69 69 69 69 69 69       |iiiiiiiiiiiiii  |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ: 9B
          +-------------------------------------------------+
          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
 +--------+-------------------------------------------------+----------------+
 |00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a                      |jjjjjjjjj       |
 +--------+-------------------------------------------------+----------------+
 14:37:10 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x744f2b47, L:/192.168.0.103:9090 - R:/192.168.0.103:49979] READ COMPLETE


二、协议与解析

1. 为什么需要协议

TCP/IP 中消息传输基于流的方式,没有边界。
协议的目的就是划定消息的边界,制定通信双方要共同遵守的通信规则

例如:在网络上传输

 下雨天留客天留我不留

是中文一句著名的无标点符号句子,在没有标点符号情况下,这句话有数种拆解方式,而意思却是完全不同,所以常被用作讲述标点符号的重要性
一种解读

 下雨天留客,天留,我不留

另一种解读

 下雨天,留客天,留我不?留

如何设计协议呢?其实就是给网络传输的信息加上“标点符号”。但通过分隔符来断句不是很好,因为分隔符本身如果用于传输,那么必须加以区分。因此,下面一种协议较为常用

 定长字节表示内容长度 + 实际内容

例如,假设一个中文字符长度为 3,按照上述协议的规则,发送信息方式如下,就不会被接收方弄错意思了

 0f下雨天留客06天留09我不留

小故事: 很久很久以前,一位私塾先生到一家任教。双方签订了一纸协议:“无鸡鸭亦可无鱼肉亦可白菜豆腐不可少不得束修金”。此后,私塾先生虽然认真教课,但主人家则总是给私塾先生以白菜豆腐为菜,丝毫未见鸡鸭鱼肉的款待。私塾先生先是很不解,可是后来也就想通了:主人把鸡鸭鱼肉的钱都会换为束修金的,也罢。至此双方相安无事。 年关将至,一个学年段亦告结束。私塾先生临行时,也不见主人家为他交付束修金,遂与主家理论。然主家亦振振有词:“有协议为证——无鸡鸭亦可,无鱼肉亦可,白菜豆腐不可少,不得束修金。这白纸黑字明摆着的,你有什么要说的呢?” 私塾先生据理力争:“协议是这样的——无鸡,鸭亦可;无鱼,肉亦可;白菜豆腐不可,少不得束修金。” 双方唇枪舌战,你来我往,真个是不亦乐乎! 这里的束修金,也作“束脩”,应当是泛指教师应当得到的报酬

2. Redis协议举例

@Slf4j
public class RedisAgreementClient {
    public static void main(String[] args) {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        byte[] LINE = {13, 10};
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast(new LoggingHandler());
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        // 会在连接 channel 建立成功后,会触发 active 事件
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) {
                            set(ctx);
                            get(ctx);
                        }
                        private void get(ChannelHandlerContext ctx) {
                            ByteBuf buf = ctx.alloc().buffer();
                            buf.writeBytes("*2".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("$3".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("get".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("$3".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("aaa".getBytes());
                            buf.writeBytes(LINE);
                            ctx.writeAndFlush(buf);
                        }
                        private void set(ChannelHandlerContext ctx) {
                            ByteBuf buf = ctx.alloc().buffer();
                            buf.writeBytes("*3".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("$3".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("set".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("$3".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("aaa".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("$3".getBytes());
                            buf.writeBytes(LINE);
                            buf.writeBytes("bbb".getBytes());
                            buf.writeBytes(LINE);
                            ctx.writeAndFlush(buf);
                        }

                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            ByteBuf buf = (ByteBuf) msg;
                            System.out.println(buf.toString(Charset.defaultCharset()));
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("localhost", 6379).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

开启客户端后查看Redis中的key
image.png

3. Http协议举例

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;

@Slf4j
public class HttpServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LoggingHandler(LogLevel.DEBUG));
                        pipeline.addLast(new HttpServerCodec());
                        //SimpleChannelInboundHandler<HttpRequest>:仅处理请求行、请求头中的内容
                        pipeline.addLast(new SimpleChannelInboundHandler<HttpRequest>() {
                            /*@Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpContent msg) throws Exception {
                                System.out.println(msg.content());
                            }*/
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
                                log.debug(msg.getClass().toString());
                                log.debug(msg.uri());

                                DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
                                byte[] bytes = "<h1>Hello World</h1>".getBytes(StandardCharsets.UTF_8);
                                response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, bytes.length);
                                response.content().writeBytes(bytes);

                                ctx.writeAndFlush(response);
                            }
                        });
                        pipeline.addLast(new SimpleChannelInboundHandler<HttpContent>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpContent msg) throws Exception {
                                System.out.println(msg.content());
                            }
                        });
                    }
                })
                .bind("localhost", 8080);
    }
}

向服务端发起请求
image.png

查看控制台打印

15:06:34 [DEBUG] [nioEventLoopGroup-2-5] i.n.h.l.LoggingHandler - [id: 0xe4b33518, L:/127.0.0.1:8080 - R:/127.0.0.1:10557] READ: 1002B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 47 45 54 20 2f 69 6e 64 65 78 20 48 54 54 50 2f |GET /index HTTP/|
|00000010| 31 2e 31 0d 0a 78 2d 61 75 74 68 2d 74 6f 6b 65 |1.1..x-auth-toke|
|00000020| 6e 3a 20 65 79 4a 68 62 47 63 69 4f 69 4a 49 55 |n: eyJhbGciOiJIU|
|00000030| 7a 49 31 4e 69 4a 39 2e 65 79 4a 7a 64 57 49 69 |zI1NiJ9.eyJzdWIi|
|00000040| 4f 69 4a 48 54 55 59 69 4c 43 4a 68 64 58 52 6f |OiJHTUYiLCJhdXRo|
|00000050| 49 6a 70 62 65 79 4a 79 62 32 78 6c 53 57 51 69 |IjpbeyJyb2xlSWQi|
|00000060| 4f 6a 45 73 49 6e 4a 76 62 47 56 4f 59 57 31 6c |OjEsInJvbGVOYW1l|
|00000070| 49 6a 6f 69 35 37 4f 37 35 37 75 66 35 36 36 68 |Ijoi57O757uf566h|
|00000080| 35 35 43 47 35 5a 47 59 49 69 77 69 63 6d 39 73 |55CG5ZGYIiwicm9s|
|00000090| 5a 55 52 6c 63 32 4d 69 4f 6d 35 31 62 47 77 73 |ZURlc2MiOm51bGws|
|000000a0| 49 6e 4a 76 62 47 56 47 62 47 46 6e 49 6a 70 75 |InJvbGVGbGFnIjpu|
|000000b0| 64 57 78 73 4c 43 4a 31 63 32 56 79 63 79 49 36 |dWxsLCJ1c2VycyI6|
|000000c0| 62 6e 56 73 62 43 77 69 59 58 56 30 61 47 39 79 |bnVsbCwiYXV0aG9y|
|000000d0| 61 58 52 70 5a 58 4d 69 4f 6d 35 31 62 47 77 73 |aXRpZXMiOm51bGws|
|000000e0| 49 6d 46 31 64 47 68 76 63 6d 6c 30 65 53 49 36 |ImF1dGhvcml0eSI6|
|000000f0| 49 75 65 7a 75 2d 65 37 6e 2d 65 75 6f 65 65 51 |Iuezu-e7n-euoeeQ|
|00000100| 68 75 57 52 6d 43 4a 39 58 53 77 69 64 58 4e 79 |huWRmCJ9XSwidXNy|
|00000110| 54 6d 46 74 5a 53 49 36 49 75 53 37 73 75 65 55 |TmFtZSI6IuS7sueU|
|00000120| 71 79 49 73 49 6e 56 7a 63 6b 46 32 59 58 52 68 |qyIsInVzckF2YXRh|
|00000130| 63 69 49 36 49 69 39 68 63 47 6b 76 61 57 31 6e |ciI6Ii9hcGkvaW1n|
|00000140| 4c 32 46 32 59 58 52 68 63 69 39 6f 5a 57 46 6b |L2F2YXRhci9oZWFk|
|00000150| 5a 58 4a 4a 62 57 63 78 4c 6d 70 77 5a 79 49 73 |ZXJJbWcxLmpwZyIs|
|00000160| 49 6e 56 7a 63 6b 6c 6b 49 6a 6f 78 4c 43 4a 6a |InVzcklkIjoxLCJj|
|00000170| 63 6d 56 68 64 47 56 45 59 58 52 6c 49 6a 6f 78 |cmVhdGVEYXRlIjox|
|00000180| 4e 6a 4d 78 4d 44 41 7a 4e 6a 67 31 4f 44 67 33 |NjMxMDAzNjg1ODg3|
|00000190| 4c 43 4a 6c 65 48 42 70 63 6d 46 30 61 57 39 75 |LCJleHBpcmF0aW9u|
|000001a0| 52 47 46 30 5a 53 49 36 4d 54 59 7a 4d 54 41 77 |RGF0ZSI6MTYzMTAw|
|000001b0| 4d 7a 6b 34 4e 54 67 34 4e 79 77 69 61 57 46 30 |Mzk4NTg4NywiaWF0|
|000001c0| 49 6a 6f 78 4e 6a 4d 78 4d 44 41 7a 4e 6a 67 31 |IjoxNjMxMDAzNjg1|
|000001d0| 4c 43 4a 6c 65 48 41 69 4f 6a 45 32 4d 7a 45 77 |LCJleHAiOjE2MzEw|
|000001e0| 4f 54 41 77 4f 44 56 39 2e 5f 4a 53 2d 75 56 36 |OTAwODV9._JS-uV6|
|000001f0| 41 46 34 43 49 72 37 66 37 45 77 4b 35 63 43 31 |AF4CIr7f7EwK5cC1|
|00000200| 31 34 6d 71 63 4f 30 5f 31 6e 47 4e 68 74 69 61 |14mqcO0_1nGNhtia|
|00000210| 4e 7a 57 45 0d 0a 55 73 65 72 2d 41 67 65 6e 74 |NzWE..User-Agent|
|00000220| 3a 20 50 6f 73 74 6d 61 6e 52 75 6e 74 69 6d 65 |: PostmanRuntime|
|00000230| 2f 37 2e 32 38 2e 34 0d 0a 41 63 63 65 70 74 3a |/7.28.4..Accept:|
|00000240| 20 2a 2f 2a 0d 0a 50 6f 73 74 6d 61 6e 2d 54 6f | */*..Postman-To|
|00000250| 6b 65 6e 3a 20 31 62 37 62 30 39 31 64 2d 65 31 |ken: 1b7b091d-e1|
|00000260| 31 62 2d 34 35 36 66 2d 39 33 34 61 2d 63 36 30 |1b-456f-934a-c60|
|00000270| 36 65 63 35 66 39 66 37 39 0d 0a 48 6f 73 74 3a |6ec5f9f79..Host:|
|00000280| 20 6c 6f 63 61 6c 68 6f 73 74 3a 38 30 38 30 0d | localhost:8080.|
|00000290| 0a 41 63 63 65 70 74 2d 45 6e 63 6f 64 69 6e 67 |.Accept-Encoding|
|000002a0| 3a 20 67 7a 69 70 2c 20 64 65 66 6c 61 74 65 2c |: gzip, deflate,|
|000002b0| 20 62 72 0d 0a 43 6f 6e 6e 65 63 74 69 6f 6e 3a | br..Connection:|
|000002c0| 20 6b 65 65 70 2d 61 6c 69 76 65 0d 0a 43 6f 6e | keep-alive..Con|
|000002d0| 74 65 6e 74 2d 54 79 70 65 3a 20 6d 75 6c 74 69 |tent-Type: multi|
|000002e0| 70 61 72 74 2f 66 6f 72 6d 2d 64 61 74 61 3b 20 |part/form-data; |
|000002f0| 62 6f 75 6e 64 61 72 79 3d 2d 2d 2d 2d 2d 2d 2d |boundary=-------|
|00000300| 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d |----------------|
|00000310| 2d 2d 2d 32 30 33 38 34 33 35 34 37 33 33 30 39 |---2038435473309|
|00000320| 37 33 39 36 39 32 30 30 31 34 33 0d 0a 43 6f 6e |73969200143..Con|
|00000330| 74 65 6e 74 2d 4c 65 6e 67 74 68 3a 20 31 36 36 |tent-Length: 166|
|00000340| 0d 0a 0d 0a 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d |....------------|
|00000350| 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d |----------------|
|00000360| 32 30 33 38 34 33 35 34 37 33 33 30 39 37 33 39 |2038435473309739|
|00000370| 36 39 32 30 30 31 34 33 0d 0a 43 6f 6e 74 65 6e |69200143..Conten|
|00000380| 74 2d 44 69 73 70 6f 73 69 74 69 6f 6e 3a 20 66 |t-Disposition: f|
|00000390| 6f 72 6d 2d 64 61 74 61 3b 20 6e 61 6d 65 3d 22 |orm-data; name="|
|000003a0| 61 61 61 22 0d 0a 0d 0a 62 62 62 62 62 62 62 62 |aaa"....bbbbbbbb|
|000003b0| 0d 0a 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d |..--------------|
|000003c0| 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 2d 32 30 |--------------20|
|000003d0| 33 38 34 33 35 34 37 33 33 30 39 37 33 39 36 39 |3843547330973969|
|000003e0| 32 30 30 31 34 33 2d 2d 0d 0a                   |200143--..      |
+--------+-------------------------------------------------+----------------+
15:06:34 [DEBUG] [nioEventLoopGroup-2-5] c.g.n.HttpServer - class io.netty.handler.codec.http.DefaultHttpRequest
15:06:34 [DEBUG] [nioEventLoopGroup-2-5] c.g.n.HttpServer - /index
15:06:34 [DEBUG] [nioEventLoopGroup-2-5] i.n.h.l.LoggingHandler - [id: 0xe4b33518, L:/127.0.0.1:8080 - R:/127.0.0.1:10557] WRITE: 59B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d |HTTP/1.1 200 OK.|
|00000010| 0a 63 6f 6e 74 65 6e 74 2d 6c 65 6e 67 74 68 3a |.content-length:|
|00000020| 20 32 30 0d 0a 0d 0a 3c 68 31 3e 48 65 6c 6c 6f | 20....<h1>Hello|
|00000030| 20 57 6f 72 6c 64 3c 2f 68 31 3e                | World</h1>     |
+--------+-------------------------------------------------+----------------+
15:06:34 [DEBUG] [nioEventLoopGroup-2-5] i.n.h.l.LoggingHandler - [id: 0xe4b33518, L:/127.0.0.1:8080 - R:/127.0.0.1:10557] FLUSH
PooledSlicedByteBuf(ridx: 0, widx: 166, cap: 166/166, unwrapped: PooledUnsafeDirectByteBuf(ridx: 1002, widx: 1002, cap: 8192))
15:06:34 [DEBUG] [nioEventLoopGroup-2-5] i.n.h.l.LoggingHandler - [id: 0xe4b33518, L:/127.0.0.1:8080 - R:/127.0.0.1:10557] READ COMPLETE

4. 自定义协议

1. 规则

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


    2. 编解码器

    根据上面的要素,设计一个登录请求消息和登录响应消息,并使用 Netty 完成收发 ```java import com.gmf.message.Message; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageCodec; import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.nio.charset.StandardCharsets; import java.util.List;

@Slf4j public class MessageCodec extends ByteToMessageCodec { @Override protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception { //1. 魔数(解码时判断) out.writeBytes(“哥么在这给你说唱”.getBytes(StandardCharsets.UTF_8)); //2. 版本号 out.writeByte(1); //3. 0=jdk序列化 1=json … out.writeByte(0); //4. 指令类型 out.writeInt(msg.getMessageType()); //5. 请求序号 out.writeInt(msg.getSequenceId()); // 无意义,对齐填充 out.writeByte(0xff); //6. 获取内容的字节数组 ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(bos); oos.writeObject(bos); byte[] bytes = bos.toByteArray(); //7. 正文长度 out.writeByte(bytes.length); //8. 消息正文 out.writeBytes(bytes); }

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    int magicNum = in.readInt();
    byte version = in.readByte();
    byte serializerType = in.readByte();
    byte messageType = in.readByte();
    int sequenceId = in.readInt();
    in.readByte();
    int length = in.readInt();
    byte[] bytes = new byte[length];
    in.readBytes(bytes, 0, length);
    ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
    Message message = (Message) ois.readObject();
    log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
    log.debug("{}", message);
    out.add(message);
}

}


测试
```java
public static void main(String[] args) throws Exception {
    EmbeddedChannel channel = new EmbeddedChannel(
        new LoggingHandler(),
        new LengthFieldBasedFrameDecoder(1024, 12, 4, 0, 0),
        new MessageCodec());

    //encode
    LoginRequestMessage message = new LoginRequestMessage("zhangsan", "123");
    //channel.writeOutbound(message);

    //decode
    ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
    new MessageCodec().encode(null, message, buffer);
    ByteBuf s1 = buffer.slice(0, 100);
    ByteBuf s2 = buffer.slice(100, buffer.readableBytes() - 100);
    s1.retain(); // 引用计数 2
    channel.writeInbound(s1); // release 1
    channel.writeInbound(s2);
}

解读
0013.png

3. 线程间共享的Handler

在创建的多个EventLoop之间它们的Handler是否能共享使用呢
image.png

在Handler类上的注解 @Sharable 表明可被共享使用
image.png
image.png

4. 💡 什么时候可以加 @Sharable

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

    @Slf4j
    @ChannelHandler.Sharable
    /**
    * 必须和 LengthFieldBasedFrameDecoder 一起使用,确保接到的 ByteBuf 消息是完整的
    */
    public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
      @Override
      protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
          ByteBuf out = ctx.alloc().buffer();
          // 1. 4 字节的魔数
          out.writeBytes(new byte[]{1, 2, 3, 4});
          // 2. 1 字节的版本,
          out.writeByte(1);
          // 3. 1 字节的序列化方式 jdk 0 , json 1
          out.writeByte(0);
          // 4. 1 字节的指令类型
          out.writeByte(msg.getMessageType());
          // 5. 4 个字节
          out.writeInt(msg.getSequenceId());
          // 无意义,对齐填充
          out.writeByte(0xff);
          // 6. 获取内容的字节数组
          ByteArrayOutputStream bos = new ByteArrayOutputStream();
          ObjectOutputStream oos = new ObjectOutputStream(bos);
          oos.writeObject(msg);
          byte[] bytes = bos.toByteArray();
          // 7. 长度
          out.writeInt(bytes.length);
          // 8. 写入内容
          out.writeBytes(bytes);
          outList.add(out);
      }
    
      @Override
      protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
          int magicNum = in.readInt();
          byte version = in.readByte();
          byte serializerType = in.readByte();
          byte messageType = in.readByte();
          int sequenceId = in.readInt();
          in.readByte();
          int length = in.readInt();
          byte[] bytes = new byte[length];
          in.readBytes(bytes, 0, length);
          ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
          Message message = (Message) ois.readObject();
          log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
          log.debug("{}", message);
          out.add(message);
      }
    }
    

三、聊天室案例

1. 聊天室业务-介绍

/**
 * 用户管理接口
 */
public interface UserService {

    /**
     * 登录
     * @param username 用户名
     * @param password 密码
     * @return 登录成功返回 true, 否则返回 false
     */
    boolean login(String username, String password);
}
/**
 * 会话管理接口
 */
public interface Session {

    /**
     * 绑定会话
     * @param channel 哪个 channel 要绑定会话
     * @param username 会话绑定用户
     */
    void bind(Channel channel, String username);

    /**
     * 解绑会话
     * @param channel 哪个 channel 要解绑会话
     */
    void unbind(Channel channel);

    /**
     * 获取属性
     * @param channel 哪个 channel
     * @param name 属性名
     * @return 属性值
     */
    Object getAttribute(Channel channel, String name);

    /**
     * 设置属性
     * @param channel 哪个 channel
     * @param name 属性名
     * @param value 属性值
     */
    void setAttribute(Channel channel, String name, Object value);

    /**
     * 根据用户名获取 channel
     * @param username 用户名
     * @return channel
     */
    Channel getChannel(String username);
}
/**
 * 聊天组会话管理接口
 */
public interface GroupSession {

    /**
     * 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null
     * @param name 组名
     * @param members 成员
     * @return 成功时返回组对象, 失败返回 null
     */
    Group createGroup(String name, Set<String> members);

    /**
     * 加入聊天组
     * @param name 组名
     * @param member 成员名
     * @return 如果组不存在返回 null, 否则返回组对象
     */
    Group joinMember(String name, String member);

    /**
     * 移除组成员
     * @param name 组名
     * @param member 成员名
     * @return 如果组不存在返回 null, 否则返回组对象
     */
    Group removeMember(String name, String member);

    /**
     * 移除聊天组
     * @param name 组名
     * @return 如果组不存在返回 null, 否则返回组对象
     */
    Group removeGroup(String name);

    /**
     * 获取组成员
     * @param name 组名
     * @return 成员集合, 没有成员会返回 empty set
     */
    Set<String> getMembers(String name);

    /**
     * 获取组成员的 channel 集合, 只有在线的 channel 才会返回
     * @param name 组名
     * @return 成员 channel 集合
     */
    List<Channel> getMembersChannel(String name);
}

2. 聊天室业务-登录

客户端

@Slf4j
public class ChatClient {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        //计数使线程等待,0则线程向下运行
        CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
        //标记是否登陆成功
        AtomicBoolean LOGIN = new AtomicBoolean(false);
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder());
                    //ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            //开辟一个新线程处理登陆
                            new Thread(() -> {
                                Scanner scanner = new Scanner(System.in);
                                System.out.print("请输入用户名:");
                                String username = scanner.next();
                                System.out.print("请输入密码:");
                                String password = scanner.next();

                                LoginRequestMessage message = new LoginRequestMessage(username, password);
                                ctx.writeAndFlush(message);

                                try {
                                    WAIT_FOR_LOGIN.await();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }

                                if (!LOGIN.get()) {
                                    ctx.channel().close();
                                    return;
                                }

                                while (true) {
                                    System.out.println("==================================");
                                    System.out.println("send [username] [content]");
                                    System.out.println("gsend [group name] [content]");
                                    System.out.println("gcreate [group name] [m1,m2,m3...]");
                                    System.out.println("gmembers [group name]");
                                    System.out.println("gjoin [group name]");
                                    System.out.println("gquit [group name]");
                                    System.out.println("quit");
                                    System.out.println("==================================");

                                    String[] line = scanner.nextLine().split(" ");
                                    switch (line[0]){
                                        case "send":
                                            ctx.writeAndFlush(new ChatRequestMessage(username, line[1], line[2]));
                                            break;
                                        case "gsend":
                                            ctx.writeAndFlush(new GroupChatRequestMessage(username, line[1], line[2]));
                                            break;
                                        case "gcreate":
                                            Set<String> set = new HashSet<>(Arrays.asList(line[2].split(",")));
                                            set.add(username); // 加入自己
                                            ctx.writeAndFlush(new GroupCreateRequestMessage(line[1], set));
                                            break;
                                        case "gmembers":
                                            ctx.writeAndFlush(new GroupMembersRequestMessage(line[1]));
                                            break;
                                        case "gjoin":
                                            ctx.writeAndFlush(new GroupJoinRequestMessage(username, line[1]));
                                            break;
                                        case "gquit":
                                            ctx.writeAndFlush(new GroupQuitRequestMessage(username, line[1]));
                                            break;
                                        case "quit":
                                            ctx.channel().close();
                                            return;
                                    }
                                }
                            }, "login").start();
                        }

                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            log.debug("msg: {}", msg);
                            if ((msg instanceof LoginResponseMessage)) {
                                LoginResponseMessage response = (LoginResponseMessage) msg;
                                LOGIN.set(response.isSuccess());
                                // 唤醒 login 线程
                                WAIT_FOR_LOGIN.countDown();
                            }
                        }
                    });
                }
            });
            Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
            channel.closeFuture().sync();
        } catch (Exception e) {
            log.error("client error", e);
        } finally {
            group.shutdownGracefully();
        }
    }
}

服务端

@Slf4j
public class ChatServer {
    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.group(boss, worker);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder());
                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    ch.pipeline().addLast(new SimpleChannelInboundHandler<LoginRequestMessage>() {
                        @Override
                        protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
                            String username = msg.getUsername();
                            String password = msg.getPassword();
                            boolean login = UserServiceFactory.getUserService().login(username, password);
                            LoginResponseMessage message;
                            if(login) {
                                //将用户channel与用户名绑定
                                SessionFactory.getSession().bind(ctx.channel(), msg.getUsername());
                                message = new LoginResponseMessage(true, "登录成功");
                            } else {
                                message = new LoginResponseMessage(false, "用户名或密码不正确");
                            }
                            ctx.writeAndFlush(message);
                        }
                    });
                }
            });
            Channel channel = serverBootstrap.bind(8080).sync().channel();
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server error", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

3. 聊天室业务-单聊

将服务端handler的代码独立出来进行重构
LoginRequestMessageHandler.java

@Slf4j
@ChannelHandler.Sharable
public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
        UserService userService = UserServiceFactory.getUserService();
        boolean login = userService.login(msg.getUsername(), msg.getPassword());

        LoginResponseMessage response;
        if (login) {
            //将用户channel与用户名绑定
            SessionFactory.getSession().bind(ctx.channel(), msg.getUsername());
            response = new LoginResponseMessage(true, "登陆成功");
        } else {
            response = new LoginResponseMessage(false, "用户名或密码错误");
        }

        ctx.writeAndFlush(response);
    }
}

ChatRequestMessageHandler.java

@ChannelHandler.Sharable
public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {
        Channel channel = SessionFactory.getSession().getChannel(msg.getTo());
        if (channel != null) {
            //对方在线
            channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));
        }else {
            //对方不在线
            ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不存在或者不在线"));
        }
    }
}

服务端
将处理单聊的Handler加入到pipeline中

ch.pipeline().addLast(CHAT_REQUEST_HANDLER);

4. 聊天室业务-群聊

1. 创建群聊

同样是创建处理 “群聊创建” 的Handler,并添加至服务端pipeline中
GroupCreateRequestMessageHandler.java

@ChannelHandler.Sharable
public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, GroupCreateRequestMessage msg) throws Exception {
        GroupSession groupSession = GroupSessionFactory.getGroupSession();
        Group group = groupSession.createGroup(msg.getGroupName(), msg.getMembers());

        if (group == null){
            ctx.writeAndFlush(new GroupCreateResponseMessage(true, "群聊创建成功"));
            List<Channel> channels = groupSession.getMembersChannel(msg.getGroupName());
            channels.forEach(channel -> {
                channel.writeAndFlush(new GroupCreateResponseMessage(true, "您被拉进群聊:" + msg.getGroupName()));
            });
        }else {
            ctx.writeAndFlush(new GroupCreateResponseMessage(false, "该群聊已存在"));
        }
    }
}

2. 推送消息

GroupChatRequestMessageHandler.java

@ChannelHandler.Sharable
public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {
        List<Channel> channels = GroupSessionFactory.getGroupSession().getMembersChannel(msg.getGroupName());
        channels.forEach(channel -> {
            //排除对自己发送
            if (channel != ctx.channel()){
                channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));
            }
        });
    }
}

3. 加入群聊

GroupJoinRequestMessageHandler.java

@ChannelHandler.Sharable
public class GroupJoinRequestMessageHandler extends SimpleChannelInboundHandler<GroupJoinRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, GroupJoinRequestMessage msg) throws Exception {
        Group group = GroupSessionFactory.getGroupSession().joinMember(msg.getGroupName(), msg.getUsername());

        if (group == null){
            ctx.writeAndFlush(new GroupQuitResponseMessage(false, "该群聊不存在"));
        }else {
            ctx.writeAndFlush(new GroupQuitResponseMessage(true, "加入成功"));
        }
    }
}

4. 退出群聊

GroupQuitRequestMessageHandler.java

@ChannelHandler.Sharable
public class GroupQuitRequestMessageHandler extends SimpleChannelInboundHandler<GroupQuitRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, GroupQuitRequestMessage msg) throws Exception {
        Group group = GroupSessionFactory.getGroupSession().removeMember(msg.getGroupName(), msg.getUsername());
        if (group == null){
            ctx.writeAndFlush(new GroupQuitResponseMessage(false, "该群聊不存在"));
        }else {
            ctx.writeAndFlush(new GroupQuitResponseMessage(true, "退出成功"));
        }
    }
}

5. 查看群成员

GroupMembersRequestMessageHandler.java

@ChannelHandler.Sharable
public class GroupMembersRequestMessageHandler extends SimpleChannelInboundHandler<GroupMembersRequestMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, GroupMembersRequestMessage msg) throws Exception {
        Set<String> members = GroupSessionFactory.getGroupSession().getMembers(msg.getGroupName());
        ctx.writeAndFlush(new GroupMembersResponseMessage(members));
    }
}

5. 聊天室业务-退出

QuitHandler.java

@Slf4j
@ChannelHandler.Sharable
public class QuitHandler extends ChannelInboundHandlerAdapter {
    /**
     * 连接正常断开时调用该方法
     * @param ctx
     * @return void
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //取消登陆状态绑定
        SessionFactory.getSession().unbind(ctx.channel());
        //super.channelInactive(ctx);
        log.debug("连接正常断开{}", ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        SessionFactory.getSession().unbind(ctx.channel());
        log.debug("连接异常断开{}", cause.getMessage());
    }
}

6. 聊天室案例-空闲检测

1. 连接假死

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

    问题
  • 假死的连接占用的资源不能自动释放

  • 向假死的连接发送数据,得到的反馈是发送超时

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

IdleStateHandler.java 构造方法
image.png

  • observeOutput:在评估写空闲时是否应该考虑字节的消耗。默认为false
  • readerIdleTime:其状态为IdleState。当在指定的时间内没有执行读操作时,将触发READER_IDLE。指定0禁用
  • writerIdleTime:其状态为IdleState。当在指定的时间内没有执行写操作时,会触发WRITER_IDLE。指定0禁用
  • allIdleTime:其状态为IdleState。在一定时间内不读不写会触发ALL_IDLE。指定0禁用
  • unit:时间单位
// 用来判断是不是 读空闲时间过长,或 写空闲时间过长
// 5s 内如果没有收到 channel 的数据,会触发一个 IdleState#READER_IDLE 事件
ch.pipeline().addLast(new IdleStateHandler(5, 0, 0));
// ChannelDuplexHandler 可以同时作为入站和出站处理器
ch.pipeline().addLast(new ChannelDuplexHandler(){
    /**
                         * 触发特殊事件
                         */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        IdleStateEvent event = (IdleStateEvent) evt;
        //如果触发的是读空闲事件
        if (event.state() == IdleState.READER_IDLE){
            log.debug("客户端超过5秒未发送数据");
            ctx.channel().close();
        }
    }
});

2. 客户端定时心跳

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

// 用来判断是不是 读空闲时间过长,或 写空闲时间过长
// 3s 内如果没有向服务器写数据,会触发一个 IdleState#WRITER_IDLE 事件
ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
ch.pipeline().addLast(new ChannelDuplexHandler(){
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //触发一个空闲事件
        IdleStateEvent event = (IdleStateEvent) evt;
        if (event.state() == IdleState.WRITER_IDLE) {
            //log.debug("3s 没有写数据了,发送一个心跳包");
            ctx.writeAndFlush(new PingMessage());
        }
    }
});