为了满足自己业务场景的需要， 应用层之间通信需要实现各种各样的网络协议。本文记录如何设计一个高效、可扩展、易维护的自定义通信协议，以及如何使用 Netty 实现自定义的通信协议。
所谓的协议，就是通信双方事先商量好的接口“暗语”， 在 TCP 网络编程中，发送方和接收方的数据包格式都是二进制，发送方将对象转化成二进制流发送给接收方，接收方获得二进制数据后需要知道如何解析对象，所以协议是双方能够正常通行的基础。
市面上已经有不小通用的协议，例如 HTTP、 HTTPS、JSON-RPC、FTP、IMAP、Protobuf等。通用协议兼容性好，易于维护，各种异构系统间可以实现无缝对接等。如果满足业务场景及性能需求的前提下，推荐采用通用协议的方案。
在特定的场景下，需要自定义自有协议。自定义协议有以下的优点：
一个较为通用的协议示例：
/*
+---------------------------------------------------------------+
| 魔数 2byte | 协议版本号 1byte | 序列化算法 1byte | 报文类型 1byte |
+---------------------------------------------------------------+
| 状态 1byte | 保留字段 4byte | 数据长度 4byte |
+---------------------------------------------------------------+
| 数据内容 （长度不定） | 校验字段 2byte |
+---------------------------------------------------------------+
*/
魔数是通信双方协商的一个暗号，通常采用固定的几个字节表示。魔数的作用是用于服务端在接收数据时先解析出前几个固定字节做正确性对比。如果和协议中的魔数不匹配，则认为是非法数据，可以直接关闭连接或采取其他措施增强系统安全性。魔数的思想在很多场景中都有体现，如 Java Class 文件开头就存储了魔数 OxCAFEBABE，在 JVM 加载 Class 文件时首先就会验证魔数对的正确性。
为了应对业务需求的变化，可能需要对自定义协议的结构或字段进行改动。不同版本的协议对应的解析方法也是不同的。所以在生产级项目中强烈建议预留协议版本这个字段。
序列化算法字段表示发送方将对象转换成二进制流，以及接收方将接收的二进制流转换成对象的方法，如 JSON、 Hessian、Java 自带序列化等。
报文类型用于描述业务场景中存在的不同报文类型。如 RPC 框架中有请求、响应、心跳类型。IM 通讯场景中有登陆、创建群聊、发送消息、接收消息、退出群聊等类型。
长度域字段代表请求数据的长度，可以定义整个报文的长度，也可以是请求数据部分的长度。
请求数据通常为的业务对象信息序列化后的二进制流。是整个报文的主体。
状态字段用于标识请求是否正常，一般由被调用方设置。例如一次 RPC 调用失败，状态字段可被服务提供方设置为异常状态。
校验字段存放某种校验算法计算报文校验码，校验码用于验证报文的正确性。
保留字段是可选项，为了应对协议升级的可能性，可以预留若干字节的保留字段，以备不时之需。
Netty 作为一个非常优秀的网络通信框架，提供了非常丰富的编解码抽象基类来实现自定义协议。
| 类型 | 编解码基类 | 说明 |
|---|---|---|
| 常用编码器类型 | MessageToByteEncoder | 对象编码成字节流 |
| MessageToMessageEncoder | 一种消息类型编码成另外一种消息类型 | |
| 常用解码器类型 | ByteToMessageDecoder/ReplayingDecoder | 将字节流解码为消息对象 |
| MessageToMessageDecoder | 将一种消息类型解码为另外一种消息类型 |
通过抽象编码类的继承图可以看出，编码类是 ChanneOutboundHandler 的抽象类实现，具体操作的是 Outbound 出站数据。
MessageToByteEncoder 用于将对象编码成字节流，只需要实现其 encode 方法即可完成自定义编码。
MessageToByteEncoder 的核心源码片段，如下所示。
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
ByteBuf buf = null;
try {
if (acceptOutboundMessage(msg)) { // 1. 消息类型是否匹配
@SuppressWarnings("unchecked")
I cast = (I) msg;
buf = allocateBuffer(ctx, cast, preferDirect); // 2. 分配 ByteBuf 资源
try {
encode(ctx, cast, buf); // 3. 执行 encode 方法完成数据编码
} finally {
ReferenceCountUtil.release(cast);
}
if (buf.isReadable()) {
ctx.write(buf, promise); // 4. 向后传递写事件
} else {
buf.release();
ctx.write(Unpooled.EMPTY_BUFFER, promise);
}
buf = null;
} else {
ctx.write(msg, promise);
}
} catch (EncoderException e) {
throw e;
} catch (Throwable e) {
throw new EncoderException(e);
} finally {
if (buf != null) {
buf.release();
}
}
}
MessageToByteEncoder 重写了 ChanneOutboundHandler 的 write() 方法，其主要逻辑分为以下几个步骤：
acceptOutboundMessage 判断是否有匹配的消息类型，如果匹配需要执行编码流程，如果不匹配直接继续传递给下一个 ChannelOutboundHandler；
分配 ByteBuf 资源，默认使用堆外内存；
调用子类实现的 encode 方法完成数据编码，一旦消息被成功编码，会通过调用 ReferenceCountUtil.release(cast) 自动释放；
如果 ByteBuf 可读，说明已经成功编码得到数据，然后写入 ChannelHandlerContext 交到下一个节点；如果 ByteBuf 不可读，则释放 ByteBuf 资源，向下传递空的 ByteBuf 对象。
编码器实现非常简单，不需要关注拆包/粘包问题。如下例子，展示了如何将字符串类型的数据写入到 ByteBuf 实例，ByteBuf 实例将传递给 ChannelPipeline 链表中的下一个 ChannelOutboundHandler。
public class StringToByteEncoder extends MessageToByteEncoder<String> {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, String data, ByteBuf byteBuf) throws Exception {
byteBuf.writeBytes(data.getBytes());
}
}
MessageToMessageEncoder 是将一种格式的消息转换为另一种格式的消息，它的子类同样只需要实现 encode 方法。MessageToMessageEncoder 常用的实现子类有 StringEncoder、LineEncoder、Base64Encoder 等。
StringEncoder 可以直接实现 String 类型数据的编码。源码示例如下：
@Override
protected void encode(ChannelHandlerContext ctx, CharSequence msg, List<Object> out) throws Exception {
if (msg.length() == 0) {
return;
}
out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset));
}
解码类是 ChanneInboundHandler 的抽象类实现，操作的是 Inbound 入站数据。解码器的主要难度在于拆包和粘包问题，由于接收方可能没有接受到完整的消息，所以编码框架还要对入站数据做缓冲处理，直到获取到完整的消息。
ByteToMessageDecoder 类将字节流转换成对象，其定义的抽象 decode 方法：
public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {
protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
protected void decodeLast(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
if (in.isReadable()) {
decodeRemovalReentryProtection(ctx, in, out);
}
}
}
encode 方法在调用时需要传入接收的数据 ByteBuf，及用来添加编码后对象的 List。处理过程如下：
MessageToMessageDecoder 是将一种消息类型的编码成另外一种消息类型。MessageToMessageDecoder 不对数据报文继续缓存，其主要用作转换消息模型。
/*
+---------------------------------------------------------------+
| 魔数 2byte | 协议版本号 1byte | 序列化算法 1byte | 报文类型 1byte |
+---------------------------------------------------------------+
| 状态 1byte | 保留字段 4byte | 数据长度 4byte |
+---------------------------------------------------------------+
| 数据内容 （长度不定） |
+---------------------------------------------------------------+
*/
对以上的自定义报文，协议头部包含了魔数、协议版本号、数据长度等固定字段。而 ByteBuf 是否完整，需要通过消息长度 dataLength 字段来判断。自定义编码器需要重写 ByteToMessageDecoder 的 encode 方法，具体代码如下所示：
@Override
public final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 判断 ByteBuf 可读取字节
if (in.readableBytes() < 14) {
return;
}
in.markReaderIndex(); // 标记 ByteBuf 读指针位置
in.skipBytes(2); // 跳过魔数
in.skipBytes(1); // 跳过协议版本号
byte serializeType = in.readByte();
in.skipBytes(1); // 跳过报文类型
in.skipBytes(1); // 跳过状态字段
in.skipBytes(4); // 跳过保留字段
int dataLength = in.readInt();
if (in.readableBytes() < dataLength) {
in.resetReaderIndex(); // 重置 ByteBuf 读指针位置
return;
}
byte[] data = new byte[dataLength];
in.readBytes(data);
SerializeService serializeService = getSerializeServiceByType(serializeType);
Object obj = serializeService.deserialize(data);
if (obj != null) {
out.add(obj);
}
}
本文学习了协议设计的基本要素，以及如何使用 Netty 编解码器实现自定义协议。最后通过基于 Netty 抽象类实现自定义的编解码器，实战具体示例协议，加深对编解码器的理解。