Netty深入浅出Java网络编程学习笔记（二） Netty进阶应用篇

目录

四、应用

1、粘包与半包

现象分析

粘包

半包

本质

解决方案

短链接

定长解码器

行解码器

长度字段解码器——LTC

2、协议设计与解析

协议的作用

Redis协议

HTTP协议

自定义协议

组成要素

编码器与解码器

编写测试类

@Sharable注解

自定义编解码器能否使用@Sharable注解

3、在线聊天室

聊天室业务

用户登录接口

用户会话接口

群聊会话接口

整体结构

客户端代码结构

服务器代码结构

登录

客户端代码

服务器代码

运行结果

单聊

群聊

创建群聊

群聊聊天

加入群聊

退出

查看群聊成员

退出聊天室

连接假死

解决方法

四、应用

1、粘包与半包

粘包和半包问题是数据传输中比较常见的问题，所谓的粘包问题是指数据在传输时，在一条消息中读取到了另一条消息的部分数据，这种现象就叫做粘包。比如发送了两条消息，分别为“ABC”和“DEF”，那么正常情况下接收端也应该收到两条消息“ABC”和“DEF”，但接收端却收到的是“ABCD”，像这种情况就叫做粘包，半包问题是指接收端只收到了部分数据，而非完整的数据的情况就叫做半包。比如发送了一条消息是“ABC”，而接收端却收到的是“AB”和“C”两条信息，这种情况就叫做半包

只要是TCP协议的网络交互都有粘包和半包问题，因为TCP的传输是基于字节的传输方式，数据是以字节的形式进行传输的，并没有明确的边界。因此，在传输过程中，TCP没有办法直接识别数据包的边界，并且在流量控制下，TCP的字节传输还不稳定，当发送方连续发送多个数据包时，这些数据包可能会在网络传输的过程中合并或拆分，导致粘包和半包问题的出现，而UDP则没有这个问题，因为UDP的传输是基于数据报的

现象分析

粘包

现象

发送 abc def，接收 abcdef

原因

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

Nagle算法的原理如下：当TCP发送方需要发送一个小数据包时，Nagle算法会将这个数据包缓存起来，不立即发送。然后，TCP发送方会继续等待其他数据，直到以下两个条件中的任意一个满足后再发送数据：

接收到之前发送的数据的确认ACK。
发送方的发送缓冲区中的数据量达到一定的阈值（一般是MSS，即最大报文长度）。

之所以要缓存起来是因为一个TCP的请求都是要进行数据报头的添加，而IP的报头+TCP的报头 = 40字节，哪怕你只是发送了1个字节的数据，也会被封装为一个41字节的传输内容，那这样子粘包现象就很严重了，为此解决方法就是，缓存多点字节再一起发过来。然而，当发送方连续发送多个小数据包时，这些数据包可能会在网络传输的过程中被合并成一个大数据包，导致粘包问题的出现。这是因为Nagle算法本身不考虑数据包的边界，只是简单地将小数据包缓存起来，直到条件满足后发送。

半包

现象

发送 abcdef，接收 abc def

原因

应用层
接收方 ByteBuf 小于实际发送数据量
传输层-网络层
滑动窗口：假设接收方的窗口只剩了 128 bytes，发送方的报文大小是 256 bytes，这时接收方窗口中无法容纳发送方的全部报文，发送方只能先发送前 128 bytes，等待 ack 后才能发送剩余部分，这就造成了半包
数据链路层
MSS 限制：当发送的数据超过 MSS （最大报文长度）限制后，会将数据切分发送，就会造成半包

本质

发生粘包与半包现象的本质是因为 TCP 是流式协议，消息无边界

解决方案

解决方案的思路和我这篇文章的处理方式类似，可以先看一下这个大概思路https://blog.csdn.net/weixin_73077810/article/details/131843387

短链接和长连接是描述客户端与服务器之间TCP连接持续时间的概念。

短链接：短链接通常指的是一次性的临时连接。在短链接中，客户端与服务器建立连接、交换数据后，连接就会关闭。在每次通信之前，需要重新建立连接，进行握手和协商。

短链接的优点是简单、轻量，适用于临时的、低频率的通信。但在高并发或频繁通信的场景中，频繁的连接建立和关闭会增加网络开销和延迟。

长连接：长连接指的是客户端与服务器之间持久的TCP连接。在长连接中，连接一经建立，客户端和服务器可以多次、长时间地进行双向通信。在连接建立后，数据可以实时、便捷地传输。

长连接的优点是减少连接建立和断开的开销，节省网络资源，减少延迟，提高通信效率。长连接常用于需要实时交互的应用，如即时通信、实时数据传输等。

需要注意的是，长连接可能会带来一些管理上的挑战。服务器需要维护大量的长连接，消耗资源，需要适当管理连接数和超时机制，防止资源浪费和死连接问题。

短链接

客户端每次向服务器发送数据以后，就与服务器断开连接，此时的消息边界为连接建立到连接断开。这时便无需使用滑动窗口等技术来缓冲数据，则不会发生粘包现象。但如果一次性数据发送过多，接收方无法一次性容纳所有数据，还是会发生半包现象，所以短链接无法解决半包现象

客户端代码改进

修改channelActive方法


public void channelActive(ChannelHandlerContext ctx) throws Exception {
    log.debug("sending...");
    ByteBuf buffer = ctx.alloc().buffer(16);
    buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
    ctx.writeAndFlush(buffer);
    // 使用短链接，每次发送完毕后就断开连接
    ctx.channel().close();
}

定长解码器

客户端于服务器约定一个最大长度，保证客户端每次发送的数据长度都不会大于该长度。若发送数据长度不足则需要补齐至该长度

服务器接收数据时，将接收到的数据按照约定的最大长度进行拆分，即使发送过程中产生了粘包，也可以通过定长解码器将数据正确地进行拆分。服务端需要用到FixedLengthFrameDecoder对数据进行定长解码，具体使用方法如下

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

客户端代码

客户端发送数据的代码如下


// 约定最大长度为16
final int maxLength = 16;
// 被发送的数据
char c = 'a';
// 向服务器发送10个报文
for (int i = 0; i < 10; i++) {
    ByteBuf buffer = ctx.alloc().buffer(maxLength);
    // 定长byte数组，未使用部分会以0进行填充
    byte[] bytes = new byte[maxLength];
    // 生成长度为0~15的数据
    for (int j = 0; j < (int)(Math.random()*(maxLength-1)); j++) {
        bytes[j] = (byte) c;
    }
    buffer.writeBytes(bytes);
    c++;
    // 将数据发送给服务器
    ctx.writeAndFlush(buffer);
}

服务器代码

使用FixedLengthFrameDecoder对粘包数据进行拆分，该handler需要添加在LoggingHandler之前，保证数据被打印时已被拆分


// 通过定长解码器对粘包数据进行拆分
ch.pipeline().addLast(new FixedLengthFrameDecoder(16));
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));

行解码器

行解码器的是通过分隔符对数据进行拆分来解决粘包半包问题的

可以通过LineBasedFrameDecoder(int maxLength)来拆分以换行符(\n or \r\n)为分隔符的数据
可以通过DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf... delimiters)来指定通过什么分隔符来拆分数据（可以传入多个分隔符）

两种解码器都需要传入数据的最大长度，若超出最大长度，会抛出TooLongFrameException异常

以换行符 \n 为分隔符

客户端代码


// 约定最大长度为 64
final int maxLength = 64;
// 被发送的数据
char c = 'a';
for (int i = 0; i < 10; i++) {
    ByteBuf buffer = ctx.alloc().buffer(maxLength);
    // 生成长度为0~62的数据
    Random random = new Random();
    StringBuilder sb = new StringBuilder();
    for (int j = 0; j < (int)(random.nextInt(maxLength-2)); j++) {
        sb.append(c);
    }
    // 数据以 \n 结尾
    sb.append("\n");
    buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));
    c++;
    // 将数据发送给服务器
    ctx.writeAndFlush(buffer);
}

服务器代码


// 通过行解码器对粘包数据进行拆分，以 \n 为分隔符
// 需要指定最大长度
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(64));
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));

以自定义分隔符 \c 为分隔符

客户端代码


// 数据以 \c 结尾
sb.append("\\c");
buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));

服务器代码


// 将分隔符放入ByteBuf中
ByteBuf bufSet = ch.alloc().buffer().writeBytes("\\c".getBytes(StandardCharsets.UTF_8));
// 通过行解码器对粘包数据进行拆分，以 \c 为分隔符
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(64, ch.alloc().buffer().writeBytes(bufSet)));
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));

长度字段解码器——LTC

在传送数据时可以在数据中添加一个用于表示有用数据长度的字段，在解码时读取出这个用于表明长度的字段，同时读取其他相关参数，即可知道最终需要的数据是什么样子的

LengthFieldBasedFrameDecoder解码器可以提供更为丰富的拆分方法，其构造方法有五个参数


public LengthFieldBasedFrameDecoder(
    int maxFrameLength,
    int lengthFieldOffset, int lengthFieldLength,
    int lengthAdjustment, int initialBytesToStrip)

参数解析

maxFrameLength 数据最大长度
- 表示数据的最大长度（包括附加信息、长度标识等内容）
lengthFieldOffset 数据长度标识的起始偏移量
- 用于指明数据第几个字节开始是用于标识有用字节长度的，因为前面可能还有其他附加信息
lengthFieldLength 数据长度标识所占字节数（用于指明有用数据的长度）
- 数据中用于表示有用数据长度的标识所占的字节数
lengthAdjustment 长度表示为有用数据的偏移量
- 用于指明数据长度标识和有用数据之间的距离，因为两者之间还可能有附加信息
initialBytesToStrip 数据读取起点
- 读取起点，不读取 0 ~ initialBytesToStrip 之间的数据

参数图解

使用

通过 EmbeddedChannel 对 handler 进行测试


public class EncoderStudy {
    public static void main(String[] args) {
        // 模拟服务器
        // 使用EmbeddedChannel测试handler
        EmbeddedChannel channel = new EmbeddedChannel(
                // 数据最大长度为1KB，长度标识前后各有1个字节的附加信息，长度标识长度为4个字节（int）
                new LengthFieldBasedFrameDecoder(1024, 1, 4, 1, 0),
                new LoggingHandler(LogLevel.DEBUG)
        );
 
        // 模拟客户端，写入数据
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
        send(buffer, "Hello");
        channel.writeInbound(buffer);
        send(buffer, "World");
        channel.writeInbound(buffer);
    }
 
    private static void send(ByteBuf buf, String msg) {
        // 得到数据的长度
        int length = msg.length();
        byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
        // 将数据信息写入buf
        // 写入长度标识前的其他信息
        buf.writeByte(0xCA);
        // 写入数据长度标识
        buf.writeInt(length);
        // 写入长度标识后的其他信息
        buf.writeByte(0xFE);
        // 写入具体的数据
        buf.writeBytes(bytes);
    }
}

运行结果


146  [main] DEBUG io.netty.handler.logging.LoggingHandler  - [id: 0xembedded, L:embedded - R:embedded] READ: 11B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| ca 00 00 00 05 fe 48 65 6c 6c 6f                |......Hello     |
+--------+-------------------------------------------------+----------------+
 
146  [main] DEBUG io.netty.handler.logging.LoggingHandler  - [id: 0xembedded, L:embedded - R:embedded] READ: 11B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| ca 00 00 00 05 fe 57 6f 72 6c 64                |......World     |
+--------+-------------------------------------------------+----------------+

2、协议设计与解析

协议的作用

TCP/IP 中消息传输基于字节流的方式，没有边界

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

Redis协议

如果我们要向Redis服务器发送一条set name Nyima的指令，需要遵守如下协议


// 该指令一共有3部分，每条指令之后都要添加回车与换行符
*3\r\n
// 第一个指令的长度是3
$3\r\n
// 第一个指令是set指令
set\r\n
// 下面的指令以此类推
$4\r\n
name\r\n
$5\r\n
Nyima\r\n

客户端代码如下


public class RedisClient {
    static final Logger log = LoggerFactory.getLogger(StudyServer.class);
    public static void main(String[] args) {
        NioEventLoopGroup group =  new NioEventLoopGroup();
        try {
            ChannelFuture channelFuture = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            // 打印日志
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    // 回车与换行符
                                    final byte[] LINE = {'\r','\n'};
                                    // 获得ByteBuf
                                    ByteBuf buffer = ctx.alloc().buffer();
                                    // 连接建立后，向Redis中发送一条指令，注意添加回车与换行
                                    // set name Nyima
                                    buffer.writeBytes("*3".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$3".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("set".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$4".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("name".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$5".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("Nyima".getBytes());
                                    buffer.writeBytes(LINE);
                                    ctx.writeAndFlush(buffer);
                                }
 
                            });
                        }
                    })
                    .connect(new InetSocketAddress("localhost", 6379));
            channelFuture.sync();
            // 关闭channel
            channelFuture.channel().close().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭group
            group.shutdownGracefully();
        }
    }
}

Redis中查询执行结果

HTTP协议

HTTP协议在请求行请求头中都有很多的内容，自己实现较为困难，可以使用HttpServerCodec作为服务器端的解码器与编码器，来处理HTTP请求


// HttpServerCodec 中既有请求的解码器 HttpRequestDecoder 又有响应的编码器 HttpResponseEncoder
// Codec(CodeCombine) 一般代表该类既作为 编码器 又作为 解码器
public final class HttpServerCodec extends CombinedChannelDuplexHandler
        implements HttpServerUpgradeHandler.SourceCodec

服务器代码


public class HttpServer {
    static final Logger log = LoggerFactory.getLogger(StudyServer.class);
 
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        new ServerBootstrap()
                .group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        // 作为服务器，使用 HttpServerCodec 作为编码器与解码器
                        ch.pipeline().addLast(new HttpServerCodec());
                        // 服务器只处理HTTPRequest，具体的限定取决于泛型
                        ch.pipeline().addLast(new SimpleChannelInboundHandler() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) {
                                // 获得请求uri
                                log.debug(msg.uri());
 
                                // 获得完整响应，设置版本号与状态码
                                DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
                                // 设置响应内容
                                byte[] bytes = "Hello, World!
".getBytes(StandardCharsets.UTF_8);
                                // 设置响应体长度，避免浏览器一直接收响应内容
                                response.headers().setInt(CONTENT_LENGTH, bytes.length);
                                // 设置响应体
                                response.content().writeBytes(bytes);
 
                                // 写回响应
                                ctx.writeAndFlush(response);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

服务器负责处理请求并响应浏览器。所以只需要处理HTTP请求即可


// 服务器只处理HTTPRequest
ch.pipeline().addLast(new SimpleChannelInboundHandler()

获得请求后，需要返回响应给浏览器。需要创建响应对象DefaultFullHttpResponse，设置HTTP版本号及状态码，为避免浏览器获得响应后，因为获得CONTENT_LENGTH而一直空转，需要添加CONTENT_LENGTH字段，表明响应体中数据的具体长度


// 获得完整响应，设置版本号与状态码
DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
// 设置响应内容
byte[] bytes = "Hello, World!
".getBytes(StandardCharsets.UTF_8);
// 设置响应体长度，避免浏览器一直接收响应内容
response.headers().setInt(CONTENT_LENGTH, bytes.length);
// 设置响应体
response.content().writeBytes(bytes);

运行结果

浏览器

自定义协议

组成要素

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

编码器与解码器


public class MessageCodec extends ByteToMessageCodec {
 
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        // 设置魔数 4个字节
        out.writeBytes(new byte[]{'N','Y','I','M'});
        // 设置版本号 1个字节
        out.writeByte(1);
        // 设置序列化方式 1个字节
        out.writeByte(1);
        // 设置指令类型 1个字节
        out.writeByte(msg.getMessageType());
        // 设置请求序号 4个字节
        out.writeInt(msg.getSequenceId());
        // 为了补齐为2的次幂个字节，填充1个字节的数据，满足为16字节
        out.writeByte(0xff);
 
        // 获得序列化后的msg
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(msg);
        byte[] bytes = bos.toByteArray();
 
        // 获得并设置正文长度 长度用4个字节标识
        out.writeInt(bytes.length);
        // 设置消息正文
        out.writeBytes(bytes);
    }
 
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List