从零开始学习Netty - 学习笔记 -Netty入门【协议设计和解析】

2.协议设计和解析

协议
在计算机中，协议是指一组规则和约定，用于在不同的计算机系统之间进行通信和数据交换。计算机协议定义了数据传输的格式、顺序、错误检测和纠正方法，以及参与通信的各个实体的角色和责任。计算机协议可以在各种不同的层次上操作，包括物理层、数据链路层、网络层、传输层、会话层、表示层和应用层。

以下是一些常见的计算机协议：

传输层协议：例如TCP (Transmission Control Protocol) 和UDP (User Datagram Protocol)，用于在网络上可靠地传输数据。
网络层协议：例如IP (Internet Protocol)，负责在网络上寻址和路由数据包。
应用层协议：例如HTTP (Hypertext Transfer Protocol)、FTP (File Transfer Protocol)、SMTP (Simple Mail Transfer Protocol) 等，用于支持特定的应用程序和服务。
数据链路层协议：例如Ethernet、PPP (Point-to-Point Protocol) 等，用于在物理网络之间传输数据帧。

2.1.redis协议

Redis 使用一种简单而有效的文本协议进行通信，这种协议被称为 RESP（REdis Serialization Protocol）。RESP 是一种二进制安全的协议，它可以将多种类型的数据结构序列化为字节流进行传输，并且允许客户端和服务器之间进行高效的通信。

下面是 RESP 协议的一些基本规则：

简单字符串（Simple Strings）：以 “+” 开头，后面跟着字符串内容和回车换行符 “\r\n”。例如：+OK\r\n 表示一个成功的响应。
错误消息（Errors）：以 “-” 开头，后面跟着错误消息内容和回车换行符 “\r\n”。例如：-ERR unknown command 'foobar'\r\n 表示一个错误的响应。
整数（Integers）：以 “:” 开头，后面跟着整数内容和回车换行符 “\r\n”。例如：:1000\r\n 表示整数 1000。
批量字符串（Bulk Strings）：以 “$” 开头，后面跟着字符串的长度（以字节为单位）、字符串内容和回车换行符 “\r\n”。例如：$6\r\nfoobar\r\n 表示一个长度为 6 的字符串 “foobar”。
数组（Arrays）：以 “*” 开头，后面跟着数组的长度和数组的元素，每个元素都可以是任意 RESP 类型。例如：*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n 表示一个包含两个元素的数组，分别是字符串 “foo” 和字符串 “bar”。

在实际的通信中，客户端发送命令给 Redis 服务器，并等待服务器的响应。客户端发送的命令遵循 RESP 协议的格式，而服务器返回的响应也是 RESP 格式的。

这种简单而灵活的 RESP 协议使得 Redis 能够高效地处理各种数据类型和命令，并在性能和易用性之间找到了平衡。

代码

package com.hrfan.java_se_base.netty.protocol;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.lang.invoke.MethodHandles;
import java.nio.charset.Charset;/*** @author 13723* @version 1.0* 2024/3/3 0:03*/
public class TestRedisProtocol {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) {// 测试redis协议NioEventLoopGroup worker = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(worker);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel channel) throws Exception {channel.pipeline().addLast(new LoggingHandler());channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {/*** 连接一旦建立就发送命令* @param ctx* @throws Exception*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 如果redis 有密码 需要先发送一个auth命令//// 发送 AUTH 命令进行认证// String authCommand = "*2\r\n$4\r\nauth\r\n$5\r\n12345\r\n";// ByteBuf authBuffer = ctx.alloc().buffer();// authBuffer.writeBytes(authCommand.getBytes());// ctx.writeAndFlush(authBuffer);// 发送一个连接建立的命令// redis 协议是一种文本协议 以 \r\n 作为结束符 以$开头的是长度  以*开头的是数组// 例如 *3\r\n$3\r\nset\r\n$4\r\nname\r\n$8\r\nhrfan\r\n// 表示一个数组 有三个元素  第一个元素是set 第二个元素是name 第三个元素是hrfan// 也就是执行 set name hrfanString command = "*3\r\n$3\r\nset\r\n$4\r\nname\r\n$5\r\nhrfan\r\n";ByteBuf buffer = ctx.alloc().buffer();buffer.writeBytes(command.getBytes());ctx.writeAndFlush(buffer);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {super.channelRead(ctx, msg);// redis 接收到结果  肯定会返回信息 +OK\r\nByteBuf byteBuf = (ByteBuf) msg;String string = byteBuf.toString(Charset.defaultCharset());logger.info("redis 返回的结果是:{}", string);}});}});// 和redis建立连接ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6379).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {logger.error("client error !",e);}finally {worker.shutdownGracefully();}}
}

2.2.HTTP协议

HTTP（Hypertext Transfer Protocol，超文本传输协议）是一种用于传输超媒体文档（例如 HTML）的应用层协议，是互联网上数据传输的基础。

HTTP的特点：

无连接：
HTTP 协议是无连接的，即每个请求都是独立的，服务器处理完请求后即断开连接，因此每个请求需要单独建立连接和断开连接，无法复用连接，导致了额外的开销。

无状态：
HTTP 协议是无状态的，即服务器不会保存客户端的请求信息，每个请求之间没有关联，服务器无法知道当前请求与之前的请求是否相关。
为了实现状态保持，引入了 Cookie 和 Session 机制。

简单快速：
HTTP 协议基于请求-响应模型，简单易懂，通信速度较快。
由于 HTTP 协议的简单性，使得它被广泛应用于 Web 数据传输。

灵活性：
HTTP 协议允许传输任意类型的数据对象，不限于文本数据，也可以传输图片、视频、音频等多媒体数据。

无安全性：
HTTP 协议是明文传输的，数据传输过程中不对数据进行加密处理，容易被窃听、篡改或伪造，因此不适合传输敏感数据。

HTTP请求/响应的基本结构：

请求结构：
请求行：包括请求方法（GET、POST 等）、请求 URI 和 HTTP 版本号。
请求头部：包括客户端信息、请求资源信息、支持的压缩方法等。
请求正文：传输请求相关的数据。

响应结构：
状态行：包括 HTTP 版本号、状态码和状态描述。
响应头部：包括服务器信息、响应时间、响应内容类型等。
响应正文：包含响应的实际数据。

HTTP的方法（请求方式）：

GET：用于请求指定的资源。
POST：用于提交数据，常用于提交表单数据。
PUT：用于上传指定的 URI 表示的内容。
DELETE：用于删除指定的资源。
HEAD：与 GET 类似，但服务器只返回响应头部，不返回实际内容。
OPTIONS：用于请求目标资源所支持的通信选项。
TRACE：用于测试目的，回显服务器收到的请求，主要用于诊断。

HTTP状态码：

1xx（信息）：请求已接收，继续处理。
2xx（成功）：请求已成功被服务器接收、理解、并接受。
3xx（重定向）：需要客户端采取进一步的操作才能完成请求。
4xx（客户端错误）：请求包含语法错误或无法完成请求。
5xx（服务器错误）：服务器在处理请求的过程中发生了错误。

HTTP持久连接：

HTTP/1.1 引入了持久连接（Persistent Connection）机制，使得可以在同一连接上发送和接收多个 HTTP 请求和响应，减少了连接建立和断开的开销，提高了性能。

/*** @author 13723* @version 1.0* 2024/3/3 0:03*/
public class TestHttpProtocol {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) {// 测试HTTP协议NioEventLoopGroup boss = new NioEventLoopGroup();NioEventLoopGroup worker = new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.channel(NioServerSocketChannel.class);bootstrap.group(boss,worker);bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel channel) throws Exception {channel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));channel.pipeline().addLast(new HttpServerCodec());// 对编解码的请求结果进行处理channel.pipeline().addLast(new ChannelInboundHandlerAdapter(){@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {super.channelRead(ctx, msg);// 此时打开浏览器 输入localhost:8080 会看到请求的信息logger.error("获取的信息：{}",msg);}});}});// 建立和http之间的连接ChannelFuture channelFuture = bootstrap.bind(9999).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {logger.error("client error !",e);}finally {worker.shutdownGracefully();boss.shutdownGracefully();}}
}

但是这个信息是一个http请求的信息 但是http请求 会分为请求头和请求体
会默认发送两次请求 一次是请求头 一次是请求体
所以我们需要对请求头和请求体进行处理

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {super.channelRead(ctx, msg);// 此时打开浏览器 输入localhost:8080 会看到请求的信息logger.error("获取的信息：{}",msg);// 但是这个信息是一个http请求的信息 但是http请求 会分为请求头和请求体// 会默认发送两次请求 一次是请求头 一次是请求体// 所以我们需要对请求头和请求体进行处理if (msg instanceof HttpRequest){HttpRequest request = (HttpRequest) msg;logger.error("请求头：{}",request.headers());}else if (msg instanceof HttpContent){HttpContent content = (HttpContent) msg;ByteBuf buf = content.content();logger.error("请求体：{}",buf.toString(Charset.defaultCharset()));}
}

还可以使用添加指定处理器处理特定的内容

SimpleChannelInboundHandler 它可根据消息的类型进行选择处理，例如我们只关心HttpRequest类型的消息，Netty会自动帮你进行转换你不需要进行类型转换

// 对请求头和请求体进行处理 我们还可以使用SimpleChannelInboundHandler
// 它可根据消息的类型进行选择处理，例如我们只关心HttpRequest类型的消息
// 他会自动帮你进行转换 你不需要进行类型转换
channel.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpRequest httpRequest) throws Exception {logger.error("请求信息：{}",httpRequest);// 向浏览器返回响应// netty提供一个响应对象// 符合http协议的响应对象 第一个参数 时http协议的版本 第二个参数是响应的状态码DefaultFullHttpResponse response = new DefaultFullHttpResponse(httpRequest.protocolVersion(), HttpResponseStatus.OK);// 向浏览器写入一些内容byte[] bytes = "<h1>hello world</h1>".getBytes();response.content().writeBytes(bytes);// 设置响应头 否则浏览器会一直等待 告知箱体response.headers().setInt(HttpHeaderNames.CONTENT_LENGTH,bytes.length);// 设置响应头的类型response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/html;charset=utf-8");// 写入响应channelHandlerContext.writeAndFlush(response);}
});

2.3.自定义协议

自定义协议要素

魔改 (Magic Number)：
用于第一时间判定是否是有效数据包，通常是一个固定的字节序列或者数字，用来标识该数据包是符合自定义协议的。
例如，可以是一个特定的字节序列，如 0x7E 0x7E。

版本号 (Protocol Version)：
用于支持协议的升级，可以在协议中包含一个字段来表示协议的版本号。
这样可以在协议升级时识别和处理不同版本的协议。

序列化算法 (Serialization Algorithm)：
用于消息正文的序列化和反序列化，可以支持多种序列化算法，如 JSON、Protobuf、Hessian、JDK 自带的序列化等。
这样可以根据需求选择最适合的序列化算法来进行数据的编码和解码。

指令类型 (Instruction Type)：
用来表示消息的类型，与业务相关，包括登录、注册、单聊、群聊等操作。
可以使用一个字段来标识不同的指令类型，以便在接收方根据指令类型进行相应的业务处理。

请求序号 (Request Sequence Number)：
用于实现双工通信和提供异步能力，每个请求都有一个唯一的序号。
接收方在处理请求后，可以通过该序号将响应与请求进行关联。

消息正文长度和消息正文：
消息正文长度字段用于表示消息正文的长度，以便在解析消息时可以正确地读取到消息的内容。
消息正文则是实际的数据内容，根据指令类型和业务需求可以是不同格式的数据，例如文本、二进制、结构化数据等。

定义一个简单的自定义协议，协议由两部分组成：消息类型和消息内容。消息类型用一个字节表示，消息内容是一个字符串。

消息格式：消息类型字节消息类型字节消息内容长度字节消息内容长度字节消息内容消息内容

MessageType (消息类型)：一个字节，0表示心跳消息，1表示业务消息。
MessageContentLength (消息内容长度)：4个字节，表示消息内容的长度。
MessageContent (消息内容)：消息内容的字节数组。

编码器

1.自定义消息信息的枚举类型

MessageType 枚举定义了两种消息类型：心跳消息和业务消息，分别用 0 和 1 表示。

public enum MessageType {HEARTBEAT((byte) 0),BUSINESS((byte) 1);private final byte value;MessageType(byte value) {this.value = value;}public byte getValue() {return value;}public static MessageType valueOf(byte value) {for (MessageType type : values()) {if (type.value == value) {return type;}}throw new IllegalArgumentException("Invalid MessageType value: " + value);}
}

2.自定义协议消息类

MyProtocolMessage 类表示一个自定义协议消息，包括消息类型和消息内容。

public class MyProtocolMessage {private MessageType type;private String content;public MyProtocolMessage(MessageType type, String content) {this.type = type;this.content = content;}public MessageType getType() {return type;}public String getContent() {return content;}
}

3.自定义协议的编码器

继承自 Netty 的 MessageToByteEncoder 类，负责将 MyProtocolMessage 编码成字节流。
将消息类型、消息内容长度和消息内容依次写入 ByteBuf 中。

public class MyProtocolEncoder extends MessageToByteEncoder<MyProtocolMessage> {@Overrideprotected void encode(ChannelHandlerContext ctx, MyProtocolMessage msg, ByteBuf out) throws Exception {// 写入消息类型out.writeByte(msg.getType().getValue());// 获取消息内容的字节数组byte[] contentBytes = msg.getContent().getBytes(StandardCharsets.UTF_8);// 写入消息内容长度out.writeInt(contentBytes.length);// 写入消息内容out.writeBytes(contentBytes);}
}

4.自定义协议的解码器

继承自 Netty 的 ByteToMessageDecoder 类，负责将字节流解码成 MyProtocolMessage 对象。
读取字节流中的消息类型和消息内容长度，然后读取对应长度的字节流作为消息内容。
构造 MyProtocolMessage 对象并加入到解码器的输出列表中。

public class MyProtocolDecoder extends ByteToMessageDecoder {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {// 至少需要5个字节来解码if (in.readableBytes() < 5) {return;}// 标记当前读取位置in.markReaderIndex();// 读取消息类型byte messageType = in.readByte();// 读取消息内容长度int contentLength = in.readInt();// 如果可读字节数小于消息内容长度，说明消息不完整，重置读取位置if (in.readableBytes() < contentLength) {in.resetReaderIndex();return;}// 读取消息内容byte[] contentBytes = new byte[contentLength];in.readBytes(contentBytes);String content = new String(contentBytes, StandardCharsets.UTF_8);// 构造消息对象MyProtocolMessage message = new MyProtocolMessage(MessageType.valueOf(messageType), content);out.add(message);}
}

5.测试

使用 Netty 的 EmbeddedChannel 类模拟了一个通道来进行测试。
测试了编码器和解码器的正确性，包括编码后解码得到的消息与原消息是否相同。

public class MyProtocolTest {private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());public static void main(String[] args) {// 创建一个嵌入式通道，并添加编码器和解码器// 指定日志级别为DEBUG，可以看到编码后的字节EmbeddedChannel channel = new EmbeddedChannel(new MyProtocolEncoder(), new MyProtocolDecoder(),new LoggingHandler(LogLevel.DEBUG));// 构造一个业务消息MyProtocolMessage message = new MyProtocolMessage(MessageType.BUSINESS, "Hello, Netty!");// 写入消息到通道channel.writeOutbound(message);// 读取通道中的字节ByteBuf encoded = channel.readOutbound();// 打印编码后的字节logger.error("编码后的字节：Encoded Message: {}",encoded);// 写入编码后的字节到通道channel.writeInbound(encoded.retain());// 读取通道中的解码后的消息MyProtocolMessage decodedMessage = channel.readInbound();// 打印解码后的消息logger.error("解码后的字节：Decoded Message: {}",decodedMessage.getContent());// 关闭通道channel.finish();}
}