三. Netty 进阶

1. 粘包与半包

1.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. }

服务器端的某次输出,可以看到一次就接收了 160 个字节,而非分 10 次接收

  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

1.2 半包现象

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

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

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

  1. serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);

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

  1. 08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84] binding...
  2. 08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84, L:/0:0:0:0:0:0:0:0:8080] bound...
  3. 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
  4. 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
  5. 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]
  6. 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
  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 |.... |
  12. +--------+-------------------------------------------------+----------------+
  13. 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
  14. 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
  15. +-------------------------------------------------+
  16. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  17. +--------+-------------------------------------------------+----------------+
  18. |00000000| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  19. |00000010| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  20. |00000020| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  21. |00000030| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  22. |00000040| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  23. |00000050| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  24. |00000060| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  25. |00000070| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
  26. |00000080| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |............ |
  27. +--------+-------------------------------------------------+----------------+
  28. 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 实际每次读取的一般是它的整数倍

1.3 现象分析

粘包

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

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

半包

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

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

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

滑动窗口

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

  • 为了解决此问题,引入了窗口概念,窗口大小即决定了无需等待应答而可以继续发送的数据最大值
    Netty03-进阶 - 图2

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

    • 图中深色的部分即要发送的数据,高亮的部分即窗口
    • 窗口内的数据才允许被发送,当应答未到达前,窗口必须停止滑动
    • 如果 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

Netty03-进阶 - 图3

Nagle 算法

  • 即使发送一个字节,也需要加入 tcp 头和 ip 头,也就是总字节数会使用 41 bytes,非常不经济。因此为了提高网络利用率,tcp 希望尽可能发送足够大的数据,这就是 Nagle 算法产生的缘由
  • 该算法是指发送端即使还有应该发送的数据,但如果这部分数据很少的话,则进行延迟发送

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

1.4 解决方案

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

方法1,短链接

以解决粘包为例

  1. public class HelloWorldClient {
  2. static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
  3. public static void main(String[] args) {
  4. // 分 10 次发送
  5. for (int i = 0; i < 10; i++) {
  6. send();
  7. }
  8. }
  9. private static void send() {
  10. NioEventLoopGroup worker = new NioEventLoopGroup();
  11. try {
  12. Bootstrap bootstrap = new Bootstrap();
  13. bootstrap.channel(NioSocketChannel.class);
  14. bootstrap.group(worker);
  15. bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  16. @Override
  17. protected void initChannel(SocketChannel ch) throws Exception {
  18. log.debug("conneted...");
  19. ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  20. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  21. @Override
  22. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  23. log.debug("sending...");
  24. ByteBuf buffer = ctx.alloc().buffer();
  25. buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
  26. ctx.writeAndFlush(buffer);
  27. // 发完即关
  28. ctx.close();
  29. }
  30. });
  31. }
  32. });
  33. ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
  34. channelFuture.channel().closeFuture().sync();
  35. } catch (InterruptedException e) {
  36. log.error("client error", e);
  37. } finally {
  38. worker.shutdownGracefully();
  39. }
  40. }
  41. }

输出,略

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

方法2,固定长度

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

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

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

  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 LoggingHandler(LogLevel.DEBUG));
  14. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  15. @Override
  16. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  17. log.debug("sending...");
  18. // 发送内容随机的数据包
  19. Random r = new Random();
  20. char c = 'a';
  21. ByteBuf buffer = ctx.alloc().buffer();
  22. for (int i = 0; i < 10; i++) {
  23. byte[] bytes = new byte[8];
  24. for (int j = 0; j < r.nextInt(8); j++) {
  25. bytes[j] = (byte) c;
  26. }
  27. c++;
  28. buffer.writeBytes(bytes);
  29. }
  30. ctx.writeAndFlush(buffer);
  31. }
  32. });
  33. }
  34. });
  35. ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
  36. channelFuture.channel().closeFuture().sync();
  37. } catch (InterruptedException e) {
  38. log.error("client error", e);
  39. } finally {
  40. worker.shutdownGracefully();
  41. }
  42. }
  43. }

客户端输出

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

服务端输出

  1. 12:06:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xe3d9713f] binding...
  2. 12:06:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xe3d9713f, L:/192.168.0.103:9090] bound...
  3. 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
  4. 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
  5. 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]
  6. 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
  7. +-------------------------------------------------+
  8. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  9. +--------+-------------------------------------------------+----------------+
  10. |00000000| 61 61 61 61 00 00 00 00 |aaaa.... |
  11. +--------+-------------------------------------------------+----------------+
  12. 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
  13. +-------------------------------------------------+
  14. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  15. +--------+-------------------------------------------------+----------------+
  16. |00000000| 62 00 00 00 00 00 00 00 |b....... |
  17. +--------+-------------------------------------------------+----------------+
  18. 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
  19. +-------------------------------------------------+
  20. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  21. +--------+-------------------------------------------------+----------------+
  22. |00000000| 63 63 00 00 00 00 00 00 |cc...... |
  23. +--------+-------------------------------------------------+----------------+
  24. 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
  25. +-------------------------------------------------+
  26. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  27. +--------+-------------------------------------------------+----------------+
  28. |00000000| 64 00 00 00 00 00 00 00 |d....... |
  29. +--------+-------------------------------------------------+----------------+
  30. 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
  31. +-------------------------------------------------+
  32. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  33. +--------+-------------------------------------------------+----------------+
  34. |00000000| 00 00 00 00 00 00 00 00 |........ |
  35. +--------+-------------------------------------------------+----------------+
  36. 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
  37. +-------------------------------------------------+
  38. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  39. +--------+-------------------------------------------------+----------------+
  40. |00000000| 66 66 66 66 00 00 00 00 |ffff.... |
  41. +--------+-------------------------------------------------+----------------+
  42. 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
  43. +-------------------------------------------------+
  44. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  45. +--------+-------------------------------------------------+----------------+
  46. |00000000| 67 67 67 00 00 00 00 00 |ggg..... |
  47. +--------+-------------------------------------------------+----------------+
  48. 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
  49. +-------------------------------------------------+
  50. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  51. +--------+-------------------------------------------------+----------------+
  52. |00000000| 68 00 00 00 00 00 00 00 |h....... |
  53. +--------+-------------------------------------------------+----------------+
  54. 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
  55. +-------------------------------------------------+
  56. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  57. +--------+-------------------------------------------------+----------------+
  58. |00000000| 69 69 69 69 69 00 00 00 |iiiii... |
  59. +--------+-------------------------------------------------+----------------+
  60. 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
  61. +-------------------------------------------------+
  62. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  63. +--------+-------------------------------------------------+----------------+
  64. |00000000| 6a 6a 6a 6a 00 00 00 00 |jjjj.... |
  65. +--------+-------------------------------------------------+----------------+
  66. 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

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

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

方法3,固定分隔符

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

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

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

  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 LoggingHandler(LogLevel.DEBUG));
  14. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  15. @Override
  16. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  17. log.debug("sending...");
  18. Random r = new Random();
  19. char c = 'a';
  20. ByteBuf buffer = ctx.alloc().buffer();
  21. for (int i = 0; i < 10; i++) {
  22. for (int j = 1; j <= r.nextInt(16)+1; j++) {
  23. buffer.writeByte((byte) c);
  24. }
  25. buffer.writeByte(10);
  26. c++;
  27. }
  28. ctx.writeAndFlush(buffer);
  29. }
  30. });
  31. }
  32. });
  33. ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
  34. channelFuture.channel().closeFuture().sync();
  35. } catch (InterruptedException e) {
  36. log.error("client error", e);
  37. } finally {
  38. worker.shutdownGracefully();
  39. }
  40. }
  41. }

客户端输出

  1. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
  2. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755] REGISTERED
  3. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755] CONNECT: /192.168.0.103:9090
  4. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755, L:/192.168.0.103:63641 - R:/192.168.0.103:9090] ACTIVE
  5. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
  6. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755, L:/192.168.0.103:63641 - R:/192.168.0.103:9090] WRITE: 60B
  7. +-------------------------------------------------+
  8. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  9. +--------+-------------------------------------------------+----------------+
  10. |00000000| 61 0a 62 62 62 0a 63 63 63 0a 64 64 0a 65 65 65 |a.bbb.ccc.dd.eee|
  11. |00000010| 65 65 65 65 65 65 65 0a 66 66 0a 67 67 67 67 67 |eeeeeee.ff.ggggg|
  12. |00000020| 67 67 0a 68 68 68 68 0a 69 69 69 69 69 69 69 0a |gg.hhhh.iiiiiii.|
  13. |00000030| 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 0a |jjjjjjjjjjj. |
  14. +--------+-------------------------------------------------+----------------+
  15. 14:08:18 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x1282d755, L:/192.168.0.103:63641 - R:/192.168.0.103:9090] FLUSH

服务端输出

  1. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] c.i.n.HelloWorldServer - connected [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641]
  2. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 1B
  3. +-------------------------------------------------+
  4. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  5. +--------+-------------------------------------------------+----------------+
  6. |00000000| 61 |a |
  7. +--------+-------------------------------------------------+----------------+
  8. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 3B
  9. +-------------------------------------------------+
  10. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  11. +--------+-------------------------------------------------+----------------+
  12. |00000000| 62 62 62 |bbb |
  13. +--------+-------------------------------------------------+----------------+
  14. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 3B
  15. +-------------------------------------------------+
  16. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  17. +--------+-------------------------------------------------+----------------+
  18. |00000000| 63 63 63 |ccc |
  19. +--------+-------------------------------------------------+----------------+
  20. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 2B
  21. +-------------------------------------------------+
  22. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  23. +--------+-------------------------------------------------+----------------+
  24. |00000000| 64 64 |dd |
  25. +--------+-------------------------------------------------+----------------+
  26. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 10B
  27. +-------------------------------------------------+
  28. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  29. +--------+-------------------------------------------------+----------------+
  30. |00000000| 65 65 65 65 65 65 65 65 65 65 |eeeeeeeeee |
  31. +--------+-------------------------------------------------+----------------+
  32. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 2B
  33. +-------------------------------------------------+
  34. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  35. +--------+-------------------------------------------------+----------------+
  36. |00000000| 66 66 |ff |
  37. +--------+-------------------------------------------------+----------------+
  38. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 7B
  39. +-------------------------------------------------+
  40. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  41. +--------+-------------------------------------------------+----------------+
  42. |00000000| 67 67 67 67 67 67 67 |ggggggg |
  43. +--------+-------------------------------------------------+----------------+
  44. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 4B
  45. +-------------------------------------------------+
  46. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  47. +--------+-------------------------------------------------+----------------+
  48. |00000000| 68 68 68 68 |hhhh |
  49. +--------+-------------------------------------------------+----------------+
  50. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 7B
  51. +-------------------------------------------------+
  52. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  53. +--------+-------------------------------------------------+----------------+
  54. |00000000| 69 69 69 69 69 69 69 |iiiiiii |
  55. +--------+-------------------------------------------------+----------------+
  56. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ: 11B
  57. +-------------------------------------------------+
  58. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  59. +--------+-------------------------------------------------+----------------+
  60. |00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a 6a |jjjjjjjjjjj |
  61. +--------+-------------------------------------------------+----------------+
  62. 14:08:18 [DEBUG] [nioEventLoopGroup-3-5] i.n.h.l.LoggingHandler - [id: 0xa4b3be43, L:/192.168.0.103:9090 - R:/192.168.0.103:63641] READ COMPLETE

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

方法4,预设长度

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

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

客户端代码

  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 LoggingHandler(LogLevel.DEBUG));
  14. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  15. @Override
  16. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  17. log.debug("sending...");
  18. Random r = new Random();
  19. char c = 'a';
  20. ByteBuf buffer = ctx.alloc().buffer();
  21. for (int i = 0; i < 10; i++) {
  22. byte length = (byte) (r.nextInt(16) + 1);
  23. // 先写入长度
  24. buffer.writeByte(length);
  25. // 再
  26. for (int j = 1; j <= length; j++) {
  27. buffer.writeByte((byte) c);
  28. }
  29. c++;
  30. }
  31. ctx.writeAndFlush(buffer);
  32. }
  33. });
  34. }
  35. });
  36. ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 9090).sync();
  37. channelFuture.channel().closeFuture().sync();
  38. } catch (InterruptedException e) {
  39. log.error("client error", e);
  40. } finally {
  41. worker.shutdownGracefully();
  42. }
  43. }
  44. }

客户端输出

  1. 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - connetted...
  2. 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8] REGISTERED
  3. 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xf0f347b8] CONNECT: /192.168.0.103:9090
  4. 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
  5. 14:37:10 [DEBUG] [nioEventLoopGroup-2-1] c.i.n.HelloWorldClient - sending...
  6. 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
  7. +-------------------------------------------------+
  8. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  9. +--------+-------------------------------------------------+----------------+
  10. |00000000| 09 61 61 61 61 61 61 61 61 61 09 62 62 62 62 62 |.aaaaaaaaa.bbbbb|
  11. |00000010| 62 62 62 62 06 63 63 63 63 63 63 08 64 64 64 64 |bbbb.cccccc.dddd|
  12. |00000020| 64 64 64 64 0f 65 65 65 65 65 65 65 65 65 65 65 |dddd.eeeeeeeeeee|
  13. |00000030| 65 65 65 65 0d 66 66 66 66 66 66 66 66 66 66 66 |eeee.fffffffffff|
  14. |00000040| 66 66 02 67 67 02 68 68 0e 69 69 69 69 69 69 69 |ff.gg.hh.iiiiiii|
  15. |00000050| 69 69 69 69 69 69 69 09 6a 6a 6a 6a 6a 6a 6a 6a |iiiiiii.jjjjjjjj|
  16. |00000060| 6a |j |
  17. +--------+-------------------------------------------------+----------------+
  18. 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

服务端输出

  1. 14:36:50 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xdff439d3] binding...
  2. 14:36:51 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0xdff439d3, L:/192.168.0.103:9090] bound...
  3. 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
  4. 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
  5. 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]
  6. 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
  7. +-------------------------------------------------+
  8. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  9. +--------+-------------------------------------------------+----------------+
  10. |00000000| 61 61 61 61 61 61 61 61 61 |aaaaaaaaa |
  11. +--------+-------------------------------------------------+----------------+
  12. 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
  13. +-------------------------------------------------+
  14. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  15. +--------+-------------------------------------------------+----------------+
  16. |00000000| 62 62 62 62 62 62 62 62 62 |bbbbbbbbb |
  17. +--------+-------------------------------------------------+----------------+
  18. 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
  19. +-------------------------------------------------+
  20. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  21. +--------+-------------------------------------------------+----------------+
  22. |00000000| 63 63 63 63 63 63 |cccccc |
  23. +--------+-------------------------------------------------+----------------+
  24. 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
  25. +-------------------------------------------------+
  26. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  27. +--------+-------------------------------------------------+----------------+
  28. |00000000| 64 64 64 64 64 64 64 64 |dddddddd |
  29. +--------+-------------------------------------------------+----------------+
  30. 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
  31. +-------------------------------------------------+
  32. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  33. +--------+-------------------------------------------------+----------------+
  34. |00000000| 65 65 65 65 65 65 65 65 65 65 65 65 65 65 65 |eeeeeeeeeeeeeee |
  35. +--------+-------------------------------------------------+----------------+
  36. 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
  37. +-------------------------------------------------+
  38. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  39. +--------+-------------------------------------------------+----------------+
  40. |00000000| 66 66 66 66 66 66 66 66 66 66 66 66 66 |fffffffffffff |
  41. +--------+-------------------------------------------------+----------------+
  42. 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
  43. +-------------------------------------------------+
  44. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  45. +--------+-------------------------------------------------+----------------+
  46. |00000000| 67 67 |gg |
  47. +--------+-------------------------------------------------+----------------+
  48. 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
  49. +-------------------------------------------------+
  50. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  51. +--------+-------------------------------------------------+----------------+
  52. |00000000| 68 68 |hh |
  53. +--------+-------------------------------------------------+----------------+
  54. 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
  55. +-------------------------------------------------+
  56. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  57. +--------+-------------------------------------------------+----------------+
  58. |00000000| 69 69 69 69 69 69 69 69 69 69 69 69 69 69 |iiiiiiiiiiiiii |
  59. +--------+-------------------------------------------------+----------------+
  60. 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
  61. +-------------------------------------------------+
  62. | 0 1 2 3 4 5 6 7 8 9 a b c d e f |
  63. +--------+-------------------------------------------------+----------------+
  64. |00000000| 6a 6a 6a 6a 6a 6a 6a 6a 6a |jjjjjjjjj |
  65. +--------+-------------------------------------------------+----------------+
  66. 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

2. 协议设计与解析

2.1 为什么需要协议?

TCP/IP 中消息传输基于流的方式,没有边界。

协议的目的就是划定消息的边界,制定通信双方要共同遵守的通信规则

例如:在网络上传输

  1. 下雨天留客天留我不留

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

一种解读

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

另一种解读

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

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

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

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

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

小故事

很久很久以前,一位私塾先生到一家任教。双方签订了一纸协议:“无鸡鸭亦可无鱼肉亦可白菜豆腐不可少不得束修金”。此后,私塾先生虽然认真教课,但主人家则总是给私塾先生以白菜豆腐为菜,丝毫未见鸡鸭鱼肉的款待。私塾先生先是很不解,可是后来也就想通了:主人把鸡鸭鱼肉的钱都会换为束修金的,也罢。至此双方相安无事。

年关将至,一个学年段亦告结束。私塾先生临行时,也不见主人家为他交付束修金,遂与主家理论。然主家亦振振有词:“有协议为证——无鸡鸭亦可,无鱼肉亦可,白菜豆腐不可少,不得束修金。这白纸黑字明摆着的,你有什么要说的呢?”

私塾先生据理力争:“协议是这样的——无鸡,鸭亦可;无鱼,肉亦可;白菜豆腐不可,少不得束修金。”

双方唇枪舌战,你来我往,真个是不亦乐乎!

这里的束修金,也作“束脩”,应当是泛指教师应当得到的报酬

2.2 redis 协议举例

  1. NioEventLoopGroup worker = new NioEventLoopGroup();
  2. byte[] LINE = {13, 10};
  3. try {
  4. Bootstrap bootstrap = new Bootstrap();
  5. bootstrap.channel(NioSocketChannel.class);
  6. bootstrap.group(worker);
  7. bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  8. @Override
  9. protected void initChannel(SocketChannel ch) {
  10. ch.pipeline().addLast(new LoggingHandler());
  11. ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  12. // 会在连接 channel 建立成功后,会触发 active 事件
  13. @Override
  14. public void channelActive(ChannelHandlerContext ctx) {
  15. set(ctx);
  16. get(ctx);
  17. }
  18. private void get(ChannelHandlerContext ctx) {
  19. ByteBuf buf = ctx.alloc().buffer();
  20. buf.writeBytes("*2".getBytes());
  21. buf.writeBytes(LINE);
  22. buf.writeBytes("$3".getBytes());
  23. buf.writeBytes(LINE);
  24. buf.writeBytes("get".getBytes());
  25. buf.writeBytes(LINE);
  26. buf.writeBytes("$3".getBytes());
  27. buf.writeBytes(LINE);
  28. buf.writeBytes("aaa".getBytes());
  29. buf.writeBytes(LINE);
  30. ctx.writeAndFlush(buf);
  31. }
  32. private void set(ChannelHandlerContext ctx) {
  33. ByteBuf buf = ctx.alloc().buffer();
  34. buf.writeBytes("*3".getBytes());
  35. buf.writeBytes(LINE);
  36. buf.writeBytes("$3".getBytes());
  37. buf.writeBytes(LINE);
  38. buf.writeBytes("set".getBytes());
  39. buf.writeBytes(LINE);
  40. buf.writeBytes("$3".getBytes());
  41. buf.writeBytes(LINE);
  42. buf.writeBytes("aaa".getBytes());
  43. buf.writeBytes(LINE);
  44. buf.writeBytes("$3".getBytes());
  45. buf.writeBytes(LINE);
  46. buf.writeBytes("bbb".getBytes());
  47. buf.writeBytes(LINE);
  48. ctx.writeAndFlush(buf);
  49. }
  50. @Override
  51. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  52. ByteBuf buf = (ByteBuf) msg;
  53. System.out.println(buf.toString(Charset.defaultCharset()));
  54. }
  55. });
  56. }
  57. });
  58. ChannelFuture channelFuture = bootstrap.connect("localhost", 6379).sync();
  59. channelFuture.channel().closeFuture().sync();
  60. } catch (InterruptedException e) {
  61. log.error("client error", e);
  62. } finally {
  63. worker.shutdownGracefully();
  64. }

2.3 http 协议举例

  1. NioEventLoopGroup boss = new NioEventLoopGroup();
  2. NioEventLoopGroup worker = new NioEventLoopGroup();
  3. try {
  4. ServerBootstrap serverBootstrap = new ServerBootstrap();
  5. serverBootstrap.channel(NioServerSocketChannel.class);
  6. serverBootstrap.group(boss, worker);
  7. serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  8. @Override
  9. protected void initChannel(SocketChannel ch) throws Exception {
  10. ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  11. ch.pipeline().addLast(new HttpServerCodec());
  12. ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
  13. @Override
  14. protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
  15. // 获取请求
  16. log.debug(msg.uri());
  17. // 返回响应
  18. DefaultFullHttpResponse response =
  19. new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
  20. byte[] bytes = "<h1>Hello, world!</h1>".getBytes();
  21. response.headers().setInt(CONTENT_LENGTH, bytes.length);
  22. response.content().writeBytes(bytes);
  23. // 写回响应
  24. ctx.writeAndFlush(response);
  25. }
  26. });
  27. /*ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  28. @Override
  29. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  30. log.debug("{}", msg.getClass());
  31. if (msg instanceof HttpRequest) { // 请求行,请求头
  32. } else if (msg instanceof HttpContent) { //请求体
  33. }
  34. }
  35. });*/
  36. }
  37. });
  38. ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
  39. channelFuture.channel().closeFuture().sync();
  40. } catch (InterruptedException e) {
  41. log.error("server error", e);
  42. } finally {
  43. boss.shutdownGracefully();
  44. worker.shutdownGracefully();
  45. }

2.4 自定义协议要素

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

编解码器

根据上面的要素,设计一个登录请求消息和登录响应消息,并使用 Netty 完成收发

  1. @Slf4j
  2. public class MessageCodec extends ByteToMessageCodec<Message> {
  3. @Override
  4. protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
  5. // 1. 4 字节的魔数
  6. out.writeBytes(new byte[]{1, 2, 3, 4});
  7. // 2. 1 字节的版本,
  8. out.writeByte(1);
  9. // 3. 1 字节的序列化方式 jdk 0 , json 1
  10. out.writeByte(0);
  11. // 4. 1 字节的指令类型
  12. out.writeByte(msg.getMessageType());
  13. // 5. 4 个字节
  14. out.writeInt(msg.getSequenceId());
  15. // 无意义,对齐填充
  16. out.writeByte(0xff);
  17. // 6. 获取内容的字节数组
  18. ByteArrayOutputStream bos = new ByteArrayOutputStream();
  19. ObjectOutputStream oos = new ObjectOutputStream(bos);
  20. oos.writeObject(msg);
  21. byte[] bytes = bos.toByteArray();
  22. // 7. 长度
  23. out.writeInt(bytes.length);
  24. // 8. 写入内容
  25. out.writeBytes(bytes);
  26. }
  27. @Override
  28. protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  29. int magicNum = in.readInt();
  30. byte version = in.readByte();
  31. byte serializerType = in.readByte();
  32. byte messageType = in.readByte();
  33. int sequenceId = in.readInt();
  34. in.readByte();
  35. int length = in.readInt();
  36. byte[] bytes = new byte[length];
  37. in.readBytes(bytes, 0, length);
  38. ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
  39. Message message = (Message) ois.readObject();
  40. log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
  41. log.debug("{}", message);
  42. out.add(message);
  43. }
  44. }

测试

  1. EmbeddedChannel channel = new EmbeddedChannel(
  2. new LoggingHandler(),
  3. new LengthFieldBasedFrameDecoder(
  4. 1024, 12, 4, 0, 0),
  5. new MessageCodec()
  6. );
  7. // encode
  8. LoginRequestMessage message = new LoginRequestMessage("zhangsan", "123", "张三");
  9. // channel.writeOutbound(message);
  10. // decode
  11. ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
  12. new MessageCodec().encode(null, message, buf);
  13. ByteBuf s1 = buf.slice(0, 100);
  14. ByteBuf s2 = buf.slice(100, buf.readableBytes() - 100);
  15. s1.retain(); // 引用计数 2
  16. channel.writeInbound(s1); // release 1
  17. channel.writeInbound(s2);

解读

Netty03-进阶 - 图4

💡 什么时候可以加 @Sharable

  • 当 handler 不保存状态时,就可以安全地在多线程下被共享
  • 但要注意对于编解码器类,不能继承 ByteToMessageCodec 或 CombinedChannelDuplexHandler 父类,他们的构造方法对 @Sharable 有限制
  • 如果能确保编解码器不会保存状态,可以继承 MessageToMessageCodec 父类
  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. /**
  4. * 必须和 LengthFieldBasedFrameDecoder 一起使用,确保接到的 ByteBuf 消息是完整的
  5. */
  6. public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
  7. @Override
  8. protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
  9. ByteBuf out = ctx.alloc().buffer();
  10. // 1. 4 字节的魔数
  11. out.writeBytes(new byte[]{1, 2, 3, 4});
  12. // 2. 1 字节的版本,
  13. out.writeByte(1);
  14. // 3. 1 字节的序列化方式 jdk 0 , json 1
  15. out.writeByte(0);
  16. // 4. 1 字节的指令类型
  17. out.writeByte(msg.getMessageType());
  18. // 5. 4 个字节
  19. out.writeInt(msg.getSequenceId());
  20. // 无意义,对齐填充
  21. out.writeByte(0xff);
  22. // 6. 获取内容的字节数组
  23. ByteArrayOutputStream bos = new ByteArrayOutputStream();
  24. ObjectOutputStream oos = new ObjectOutputStream(bos);
  25. oos.writeObject(msg);
  26. byte[] bytes = bos.toByteArray();
  27. // 7. 长度
  28. out.writeInt(bytes.length);
  29. // 8. 写入内容
  30. out.writeBytes(bytes);
  31. outList.add(out);
  32. }
  33. @Override
  34. protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  35. int magicNum = in.readInt();
  36. byte version = in.readByte();
  37. byte serializerType = in.readByte();
  38. byte messageType = in.readByte();
  39. int sequenceId = in.readInt();
  40. in.readByte();
  41. int length = in.readInt();
  42. byte[] bytes = new byte[length];
  43. in.readBytes(bytes, 0, length);
  44. ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
  45. Message message = (Message) ois.readObject();
  46. log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
  47. log.debug("{}", message);
  48. out.add(message);
  49. }
  50. }

3. 聊天室案例

3.1 聊天室业务介绍

  1. /**
  2. * 用户管理接口
  3. */
  4. public interface UserService {
  5. /**
  6. * 登录
  7. * @param username 用户名
  8. * @param password 密码
  9. * @return 登录成功返回 true, 否则返回 false
  10. */
  11. boolean login(String username, String password);
  12. }
  1. /**
  2. * 会话管理接口
  3. */
  4. public interface Session {
  5. /**
  6. * 绑定会话
  7. * @param channel 哪个 channel 要绑定会话
  8. * @param username 会话绑定用户
  9. */
  10. void bind(Channel channel, String username);
  11. /**
  12. * 解绑会话
  13. * @param channel 哪个 channel 要解绑会话
  14. */
  15. void unbind(Channel channel);
  16. /**
  17. * 获取属性
  18. * @param channel 哪个 channel
  19. * @param name 属性名
  20. * @return 属性值
  21. */
  22. Object getAttribute(Channel channel, String name);
  23. /**
  24. * 设置属性
  25. * @param channel 哪个 channel
  26. * @param name 属性名
  27. * @param value 属性值
  28. */
  29. void setAttribute(Channel channel, String name, Object value);
  30. /**
  31. * 根据用户名获取 channel
  32. * @param username 用户名
  33. * @return channel
  34. */
  35. Channel getChannel(String username);
  36. }
  1. /**
  2. * 聊天组会话管理接口
  3. */
  4. public interface GroupSession {
  5. /**
  6. * 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null
  7. * @param name 组名
  8. * @param members 成员
  9. * @return 成功时返回组对象, 失败返回 null
  10. */
  11. Group createGroup(String name, Set<String> members);
  12. /**
  13. * 加入聊天组
  14. * @param name 组名
  15. * @param member 成员名
  16. * @return 如果组不存在返回 null, 否则返回组对象
  17. */
  18. Group joinMember(String name, String member);
  19. /**
  20. * 移除组成员
  21. * @param name 组名
  22. * @param member 成员名
  23. * @return 如果组不存在返回 null, 否则返回组对象
  24. */
  25. Group removeMember(String name, String member);
  26. /**
  27. * 移除聊天组
  28. * @param name 组名
  29. * @return 如果组不存在返回 null, 否则返回组对象
  30. */
  31. Group removeGroup(String name);
  32. /**
  33. * 获取组成员
  34. * @param name 组名
  35. * @return 成员集合, 没有成员会返回 empty set
  36. */
  37. Set<String> getMembers(String name);
  38. /**
  39. * 获取组成员的 channel 集合, 只有在线的 channel 才会返回
  40. * @param name 组名
  41. * @return 成员 channel 集合
  42. */
  43. List<Channel> getMembersChannel(String name);
  44. }

3.2 聊天室业务-登录

  1. @Slf4j
  2. public class ChatServer {
  3. public static void main(String[] args) {
  4. NioEventLoopGroup boss = new NioEventLoopGroup();
  5. NioEventLoopGroup worker = new NioEventLoopGroup();
  6. LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  7. MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  8. try {
  9. ServerBootstrap serverBootstrap = new ServerBootstrap();
  10. serverBootstrap.channel(NioServerSocketChannel.class);
  11. serverBootstrap.group(boss, worker);
  12. serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  13. @Override
  14. protected void initChannel(SocketChannel ch) throws Exception {
  15. ch.pipeline().addLast(new ProcotolFrameDecoder());
  16. ch.pipeline().addLast(LOGGING_HANDLER);
  17. ch.pipeline().addLast(MESSAGE_CODEC);
  18. ch.pipeline().addLast(new SimpleChannelInboundHandler<LoginRequestMessage>() {
  19. @Override
  20. protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
  21. String username = msg.getUsername();
  22. String password = msg.getPassword();
  23. boolean login = UserServiceFactory.getUserService().login(username, password);
  24. LoginResponseMessage message;
  25. if(login) {
  26. message = new LoginResponseMessage(true, "登录成功");
  27. } else {
  28. message = new LoginResponseMessage(false, "用户名或密码不正确");
  29. }
  30. ctx.writeAndFlush(message);
  31. }
  32. });
  33. }
  34. });
  35. Channel channel = serverBootstrap.bind(8080).sync().channel();
  36. channel.closeFuture().sync();
  37. } catch (InterruptedException e) {
  38. log.error("server error", e);
  39. } finally {
  40. boss.shutdownGracefully();
  41. worker.shutdownGracefully();
  42. }
  43. }
  44. }
  1. @Slf4j
  2. public class ChatClient {
  3. public static void main(String[] args) {
  4. NioEventLoopGroup group = new NioEventLoopGroup();
  5. LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
  6. MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
  7. CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
  8. AtomicBoolean LOGIN = new AtomicBoolean(false);
  9. try {
  10. Bootstrap bootstrap = new Bootstrap();
  11. bootstrap.channel(NioSocketChannel.class);
  12. bootstrap.group(group);
  13. bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  14. @Override
  15. protected void initChannel(SocketChannel ch) throws Exception {
  16. ch.pipeline().addLast(new ProcotolFrameDecoder());
  17. // ch.pipeline().addLast(LOGGING_HANDLER);
  18. ch.pipeline().addLast(MESSAGE_CODEC);
  19. ch.pipeline().addLast("client handler", new ChannelInboundHandlerAdapter() {
  20. // 接收响应消息
  21. @Override
  22. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  23. log.debug("msg: {}", msg);
  24. if ((msg instanceof LoginResponseMessage)) {
  25. LoginResponseMessage response = (LoginResponseMessage) msg;
  26. if (response.isSuccess()) {
  27. // 如果登录成功
  28. LOGIN.set(true);
  29. }
  30. // 唤醒 system in 线程
  31. WAIT_FOR_LOGIN.countDown();
  32. }
  33. }
  34. // 在连接建立后触发 active 事件
  35. @Override
  36. public void channelActive(ChannelHandlerContext ctx) throws Exception {
  37. // 负责接收用户在控制台的输入,负责向服务器发送各种消息
  38. new Thread(() -> {
  39. Scanner scanner = new Scanner(System.in);
  40. System.out.println("请输入用户名:");
  41. String username = scanner.nextLine();
  42. System.out.println("请输入密码:");
  43. String password = scanner.nextLine();
  44. // 构造消息对象
  45. LoginRequestMessage message = new LoginRequestMessage(username, password);
  46. // 发送消息
  47. ctx.writeAndFlush(message);
  48. System.out.println("等待后续操作...");
  49. try {
  50. WAIT_FOR_LOGIN.await();
  51. } catch (InterruptedException e) {
  52. e.printStackTrace();
  53. }
  54. // 如果登录失败
  55. if (!LOGIN.get()) {
  56. ctx.channel().close();
  57. return;
  58. }
  59. while (true) {
  60. System.out.println("==================================");
  61. System.out.println("send [username] [content]");
  62. System.out.println("gsend [group name] [content]");
  63. System.out.println("gcreate [group name] [m1,m2,m3...]");
  64. System.out.println("gmembers [group name]");
  65. System.out.println("gjoin [group name]");
  66. System.out.println("gquit [group name]");
  67. System.out.println("quit");
  68. System.out.println("==================================");
  69. String command = scanner.nextLine();
  70. String[] s = command.split(" ");
  71. switch (s[0]){
  72. case "send":
  73. ctx.writeAndFlush(new ChatRequestMessage(username, s[1], s[2]));
  74. break;
  75. case "gsend":
  76. ctx.writeAndFlush(new GroupChatRequestMessage(username, s[1], s[2]));
  77. break;
  78. case "gcreate":
  79. Set<String> set = new HashSet<>(Arrays.asList(s[2].split(",")));
  80. set.add(username); // 加入自己
  81. ctx.writeAndFlush(new GroupCreateRequestMessage(s[1], set));
  82. break;
  83. case "gmembers":
  84. ctx.writeAndFlush(new GroupMembersRequestMessage(s[1]));
  85. break;
  86. case "gjoin":
  87. ctx.writeAndFlush(new GroupJoinRequestMessage(username, s[1]));
  88. break;
  89. case "gquit":
  90. ctx.writeAndFlush(new GroupQuitRequestMessage(username, s[1]));
  91. break;
  92. case "quit":
  93. ctx.channel().close();
  94. return;
  95. }
  96. }
  97. }, "system in").start();
  98. }
  99. });
  100. }
  101. });
  102. Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
  103. channel.closeFuture().sync();
  104. } catch (Exception e) {
  105. log.error("client error", e);
  106. } finally {
  107. group.shutdownGracefully();
  108. }
  109. }
  110. }

3.3 聊天室业务-单聊

服务器端将 handler 独立出来

登录 handler

  1. @ChannelHandler.Sharable
  2. public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> {
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception {
  5. String username = msg.getUsername();
  6. String password = msg.getPassword();
  7. boolean login = UserServiceFactory.getUserService().login(username, password);
  8. LoginResponseMessage message;
  9. if(login) {
  10. SessionFactory.getSession().bind(ctx.channel(), username);
  11. message = new LoginResponseMessage(true, "登录成功");
  12. } else {
  13. message = new LoginResponseMessage(false, "用户名或密码不正确");
  14. }
  15. ctx.writeAndFlush(message);
  16. }
  17. }

单聊 handler

  1. @ChannelHandler.Sharable
  2. public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {
  5. String to = msg.getTo();
  6. Channel channel = SessionFactory.getSession().getChannel(to);
  7. // 在线
  8. if(channel != null) {
  9. channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));
  10. }
  11. // 不在线
  12. else {
  13. ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不存在或者不在线"));
  14. }
  15. }
  16. }

3.4 聊天室业务-群聊

创建群聊

  1. @ChannelHandler.Sharable
  2. public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, GroupCreateRequestMessage msg) throws Exception {
  5. String groupName = msg.getGroupName();
  6. Set<String> members = msg.getMembers();
  7. // 群管理器
  8. GroupSession groupSession = GroupSessionFactory.getGroupSession();
  9. Group group = groupSession.createGroup(groupName, members);
  10. if (group == null) {
  11. // 发生成功消息
  12. ctx.writeAndFlush(new GroupCreateResponseMessage(true, groupName + "创建成功"));
  13. // 发送拉群消息
  14. List<Channel> channels = groupSession.getMembersChannel(groupName);
  15. for (Channel channel : channels) {
  16. channel.writeAndFlush(new GroupCreateResponseMessage(true, "您已被拉入" + groupName));
  17. }
  18. } else {
  19. ctx.writeAndFlush(new GroupCreateResponseMessage(false, groupName + "已经存在"));
  20. }
  21. }
  22. }

群聊

  1. @ChannelHandler.Sharable
  2. public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {
  5. List<Channel> channels = GroupSessionFactory.getGroupSession()
  6. .getMembersChannel(msg.getGroupName());
  7. for (Channel channel : channels) {
  8. channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));
  9. }
  10. }
  11. }

加入群聊

  1. @ChannelHandler.Sharable
  2. public class GroupJoinRequestMessageHandler extends SimpleChannelInboundHandler<GroupJoinRequestMessage> {
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, GroupJoinRequestMessage msg) throws Exception {
  5. Group group = GroupSessionFactory.getGroupSession().joinMember(msg.getGroupName(), msg.getUsername());
  6. if (group != null) {
  7. ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群加入成功"));
  8. } else {
  9. ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
  10. }
  11. }
  12. }

退出群聊

  1. @ChannelHandler.Sharable
  2. public class GroupQuitRequestMessageHandler extends SimpleChannelInboundHandler<GroupQuitRequestMessage> {
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, GroupQuitRequestMessage msg) throws Exception {
  5. Group group = GroupSessionFactory.getGroupSession().removeMember(msg.getGroupName(), msg.getUsername());
  6. if (group != null) {
  7. ctx.writeAndFlush(new GroupJoinResponseMessage(true, "已退出群" + msg.getGroupName()));
  8. } else {
  9. ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
  10. }
  11. }
  12. }

查看成员

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

3.5 聊天室业务-退出

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class QuitHandler extends ChannelInboundHandlerAdapter {
  4. // 当连接断开时触发 inactive 事件
  5. @Override
  6. public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  7. SessionFactory.getSession().unbind(ctx.channel());
  8. log.debug("{} 已经断开", ctx.channel());
  9. }
  10. // 当出现异常时触发
  11. @Override
  12. public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  13. SessionFactory.getSession().unbind(ctx.channel());
  14. log.debug("{} 已经异常断开 异常是{}", ctx.channel(), cause.getMessage());
  15. }
  16. }

3.6 聊天室业务-空闲检测

连接假死

原因

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

问题

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

服务器端解决

  • 怎么判断客户端连接是否假死呢?如果能收到客户端数据,说明没有假死。因此策略就可以定为,每隔一段时间就检查这段时间内是否接收到客户端数据,没有就可以判定为连接假死
  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 (event.state() == IdleState.READER_IDLE) {
  12. log.debug("已经 5s 没有读到数据了");
  13. ctx.channel().close();
  14. }
  15. }
  16. });

客户端定时心跳

  • 客户端可以定时向服务器端发送数据,只要这个时间间隔小于服务器定义的空闲检测的时间间隔,那么就能防止前面提到的误判,客户端可以定义如下心跳处理器
  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.debug("3s 没有写数据了,发送一个心跳包");
  13. ctx.writeAndFlush(new PingMessage());
  14. }
  15. }
  16. });