Netty底子—8.Netty实现私有协议栈一

大纲
1.私有协议先容
2.私有协议的通讯模子
3.私有协议栈的消息界说
4.私有协议栈链路的创建
5.私有协议栈链路的关闭
6.私有协议栈的心跳机制
7.私有协议栈的重连机制
8.私有协议栈的重复登录掩护
9.私有协议栈焦点的ChannelHandler
10.私有协议栈的客户端和服务端
11.私有协议栈的Packet数据包与编解码
12.私有协议栈的会话ID处置惩罚器
13.私有协议栈的握手处置惩罚器
14.私有协议栈的链路保活处置惩罚器

1.私有协议先容
(1)什么是私有协议
(2)公有协议与私有协议
(3)利用Netty定制私有协议
(4)私有协议栈的根本功能

(1)什么是私有协议
跨节点的长途服务调用(跨节点通讯)，除了链路层的物理毗连外，还须要对哀求和相应消息举行编解码。在哀求和应答消息自己以外，也须要携带一些其他控制和管理类指令。比方链路创建的握手哀求和相应消息、链路检测的心跳消息等。当这些功能组合到一起后就会形成私有协议。私有协议并没有标准的界说，只要是可以大概用于跨进程、跨主机数据互换的非标准协议，都可以称为私有协议。

(2)公有协议与私有协议
公有协议是雷同于HTTP、WebSocket如许公开的一套协议，全部人都可以基于这个协议来举行通讯。

私有协议是雷同于RPC如许的自界说的通讯协议，须要自界说哀求消息应包罗哪些内容和相应消息应包罗哪些内容。

方向于中央件类的软件产物，内部的各节点之间通常利用私有协议举行通讯。Java Web体系、业务体系内部之间的通讯，则通常利用Dubbo这种RPC框架举行通讯。Dubbo有一套Dubbo协议，业务体系之间的通讯直接利用Dubbo协议即可，无须再自界说协议。对于其他的比如自研微服务管理平台，那么其内部通讯大概就须要自界说一套私有协议。

(3)利用Netty定制私有协议
绝大多数的私有协议的传输层都是基于TCP/IP协议，以是利用Netty的NIO TCP协议栈可以非常方便地举行私有协议的定制和开辟。

下面利用Netty提供的异步TCP协议栈开辟一个私有协议栈。这个私有协议栈可用于内部各模块之间的通讯，它基于TCP/IP协议栈，是一个类HTTP协议的应用层协议栈。相比于传统的标准协议栈，Netty协议栈更加轻巧、机动。

创建在TCP协议之上的应用层公有协议有：HTTP协议。TCP毗连创建后，发送的哀求数据时带的数据会按照HTTP协议来构造，返回的相应数据也会按照HTTP协议来构造。

(4)私有协议栈的根本功能
这个私有协议栈可用于承载业务内部各模块之间的消息交互和服务调用，功能有：
一.提供高性能的异步通讯本领(基于Netty)
二.提供消息的编码解码本领
三.提供IP白名单认证机制
四.提供链路的保活探测机制
五.提供链路的断连重连机制

2.私有协议栈的通讯模子
一.客户端发送握手哀求消息
二.服务端对握手哀求消息举行正当性校验
三.链路创建乐成后，客户端发送业务消息
四.链路创建乐成后，服务端发送心跳消息
五.链路创建乐成后，客户端发送心跳消息
六.链路创建乐成后，服务端发送业务消息

阐明一：Netty协议栈的通讯双方在链路创建乐成后，双方可举行全双工通讯。无论客户端还是服务端，都可以主动发送哀求消息给对方，通讯方可用是TWO WAY大概ONE WAY。

阐明二：双方之间的心跳接纳Ping-Pong机制。当链路处于空闲时(即链路已经长时间没有通讯)，客户端主动发送Ping消息给服务端，服务端收到Ping消息后发送应答消息(即Pong消息)给客户端。

阐明三：如果客户端一连发送N条Ping消息都没有收到服务端返回的Pong消息，阐明链路断开或服务端非常。此时客户端会主动关闭毗连，隔断周期T之后再发起重连利用，直到重连乐成。

3.私有协议栈的消息界说
任何一个自界说协议，都必须有消息头Header + 消息体Body。消息头Header里会存放一些消息的元数据，消息体Body里会存放完备的哀求体数据。

消息头Header里可以放：

一.crcCode(32位的int型)
CRC查验码由三部门构成：第一部门是2个字节的固定值，表明这是某协议的消息。第二部门是1字节的消息主版本号，第三部门是1字节的消息次版本号。

二.length(32位的int型)
这是整个消息的消息长度：包罗消息头Header + 消息体Body。

三.SessionId(64位的long型)
创建TCP长毗连后，这个长毗连就是一个会话，会话ID在集群节点内全局唯一。

四.type(8位Byte型)
表现当前消息的范例。比如0是业务哀求消息、1是业务相应消息、2是业务ONE WAY消息(既是哀求又是相应消息)、3是握手哀求消息、4是握手相应消息、5是心跳哀求消息、6是心跳应答消息。

4.私有协议栈链路的创建
如果A节点须要调用B节点的服务，但是A和B之间还没有创建物理层链路，即Netty的connect()方法触发的TCP的三次握手。那么会由调用方主动发起毗连，此时调用方为客户端，被调用方为服务端。

创建好物理层链路后，还须要创建应用层链路。此时客户端会发送握手哀求消息给服务端，服务端收到握手哀求消息后，如果通过IP白名单校验，则返回握手乐成应答消息给客户端，应用层链路创建乐成。

之后，客户端和服务端就可以相互发送业务消息了。固然，服务端还要处置惩罚握手哀求超时的情况。

5.私有协议栈链路的关闭
由于客户端和服务端接纳长毗连举行通讯，而且在正常的业务运行期间，双方通过心跳和业务消息来维持链路，以是任何一方都不须要主动关闭毗连。

但在以下情况下，客户端和服务端须要关闭毗连：
一.一方宕机或重启
二.消息读写IO非常
三.心跳消息读写IO非常
四.心跳超时
五.编码非常

6.私有协议栈的心跳机制
心跳机制用于办理网络超时、闪断、对方进程僵死大概处置惩罚迟钝等情况。客户端和服务端会在网络空闲时接纳心跳机制来检测链路的互通性，一旦发现网络故障，立即关闭链路、主动重连。

阐明一：每隔时间周期T如10秒，客户端和服务端各自做一次心跳查抄。

阐明二：当此中一方发现网络处于空闲状态的连续时间已经到达t比如1分钟时，那么就主动发送Ping心跳消息给另一方。

阐明三：如果在下一个周期T到来时，一方没有收到另一方发送的Pong心跳应答消息大概业务消息，则心跳失败计数器 + 1。

阐明四：每当一方吸收到另一方的Pong心跳应答消息大概业务消息时，则将心跳失败计数器清零。一连N次没有收到另一方的Pong心跳应答消息大概业务消息时，则关闭链路。如果是客户端关闭了链路，那么在隔断INTERVAL时间后再发起重连利用。

Ping-Pong双向心跳机制：客户端和服务端共用一个KeepAliveHandler，并各自启动心跳线程。

7.私有协议栈的重连机制
如果链路制止，客户端等待INTERVAL时间后会发起重连利用。如果重连失败，隔断INTERVAL时间后再次发起重连，直到重连乐成。

为了包管服务端能有富足的时间开释句柄资源，在初次断连时客户端须要等待INTERVAL时间后再发起重连，而不是失败后就立即重连。重连失败时，客户端也要包管自身资源能被实时开释。

8.私有协议栈的重复登录掩护
当客户端握手乐成后，在链路处于正常状态下，不允许客户端重复登录，以防止客户端在非常状态下反复重连导致服务端的句柄资源被耗尽。

服务端吸收到客户端的握手哀求消息后，起首对IP所在举行正当性查验。如果查验乐成，则在会话缓存中检察客户端是否已经登录。如果已经登录，则拒绝重复登录，同时关闭TCP链路。

客户端收到握手失败的消息后，要关闭客户端的TCP毗连，等待INTERVAL时间后，再发起TCP毗连，直到认证乐成。

当服务端一连N次心跳超时后须要主动关闭链路，同时还要清空该客户端的会话缓存信息，以包管可重连乐成。

9.私有协议栈焦点的ChannelHandler
告急五种ChannelHandler：
一.Packet数据包编码器PacketEncoder
二.Packet数据包解码器PacketDecoder
三.握手机制处置惩罚HandShakeHandler
四.会话ID管理SessionIdHandler
五.链路保活处置惩罚KeepAliveHandler

2. ch.pipeline()
3. .addLast(new PacketEncoder())
4. .addLast(new PacketDecoder(MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH))
5. .addLast(new SessionIdHandler(false))
6. .addLast(new HandshakeHandler(NettyMode.SERVER.value()))
7. .addLast(new KeepAliveHandler());

复制代码

SessionIdHandler会话ID处置惩罚器(重复登录掩护)在pipeline链条中排第三。
HandshakeHandler握手处置惩罚器(握手超时查抄)在pipeline链条排第四。
KeepAliveHandler链路保活处置惩罚器(心跳查抄机制)在pipeline链条排第五。

10.私有协议栈的客户端和服务端
(1)可断网重连的客户端

2. public class NettyClient {
3.     private static final Logger logger = LogManager.getLogger(NettyClient.class);
4.     private static final int SCHEDULED_THREAD_NUM = 1;
5.     private static final int MAX_FRAME_LENGTH = 1024 * 1024;
6.     private static final int LENGTH_FIELD_OFFSET = 4;
7.     private static final int LENGTH_FIELD_LENGTH = 4;
9.     private ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(SCHEDULED_THREAD_NUM);
10.     private EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
12.     public void connect(final String host, final int port) throws InterruptedException {
13.         try {
14.             Bootstrap bootstrap = new Bootstrap();
15.             bootstrap.group(eventLoopGroup)
16.             .channel(NioSocketChannel.class)
17.             .option(ChannelOption.TCP_NODELAY, true)
18.             .handler(new ChannelInitializer<SocketChannel>() {
19.                 protected void initChannel(SocketChannel ch) throws Exception {
20.                     ch.pipeline()
21.                     .addLast(new PacketEncoder())
22.                     .addLast(new PacketDecoder(MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH))
23.                     .addLast(new SessionIdHandler(true))
24.                     .addLast(new HandshakeHandler(NettyMode.CLIENT.value()))
25.                     .addLast(new KeepAliveHandler());
26.                 }
27.             });
28.             ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(host, port)).sync();
29.             channelFuture.channel().closeFuture().sync();//阻塞等待直到连接被关闭
30.         } finally {
31.             //如果连接被关闭了，上面的closeFuture().sync()就不会再进行阻塞，于是下面的代码就会被执行进行重连
32.             threadPool.execute(new Runnable() {
33.                 public void run() {
34.                     try {
35.                         TimeUnit.SECONDS.sleep(5);
36.                         try {
37.                             //递归调用
38.                             connect(host,  port);
39.                         } catch(InterruptedException e) {
40.                             logger.error("Netty client connect error.");
41.                         }
42.                     } catch (InterruptedException e) {
43.                         logger.error("Socket channel close thread interrupted exception.");
44.                     }
45.                 }
46.             });
47.         }
48.     }
49. }

复制代码

(2)私有协议栈的服务端

2. public class NettyServer {
3.     private static final int MAX_FRAME_LENGTH = 1024 * 1024;
4.     private static final int LENGTH_FIELD_OFFSET = 4;
5.     private static final int LENGTH_FIELD_LENGTH = 4;
6.    
7.     public void bind(int port) throws Exception {
8.         EventLoopGroup bossEventLoopGroup = new NioEventLoopGroup();
9.         EventLoopGroup workerEventLoopGroup = new NioEventLoopGroup();
11.         ServerBootstrap serverBootstrap = new ServerBootstrap();
12.         serverBootstrap.group(bossEventLoopGroup, workerEventLoopGroup)
13.         .channel(NioServerSocketChannel.class)
14.         .option(ChannelOption.SO_BACKLOG, 100)
15.         .childHandler(new ChannelInitializer<SocketChannel>() {
16.             protected void initChannel(SocketChannel ch) throws Exception {
17.                 ch.pipeline()
18.                 .addLast(new PacketEncoder())
19.                 .addLast(new PacketDecoder(MAX_FRAME_LENGTH, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH))
20.                 .addLast(new SessionIdHandler(false))
21.                 .addLast(new HandshakeHandler(NettyMode.SERVER.value()))
22.                 .addLast(new KeepAliveHandler());
23.             }
24.         });
25.         serverBootstrap.bind(port).sync();
26.     }
27. }

复制代码

11.私有协议栈的Packet数据包与编解码
(1)Packet数据包的数据结构
(2)从数据读取角度美满Packet数据包
(3)Packet数据包的编码器实现
(4)基于builder计划模式重构Packet数据包
(5)Packet数据包的解码器实现

(1)Packet数据包的数据结构

2. //数据包
3. public class Packet {
4.     private Header header = new Header();
5.     private Body body = new Body();
7.     class Header {
8.         //4个字节检验值
9.         private int crc;
10.         //完整数据包的字节数量的大小
11.         private int size;
12.         //节点之间建立的会话的id，物理链路一旦建立好了以后，就必然会分配一个唯一的sessionId，8个字节大小
13.         private long sessionId;
14.         //数据包类型，区分包是个什么，握手请求包，握手应答包
15.         private byte type;
16.         //数据包重要级别
17.         private byte level;
18.         //扩展消息头
19.         private Map<String, Object> extend = new HashMap<String, Object>();
20.     }
21.    
22.     class Body {
23.         
24.     }
25. }

复制代码

(2)从数据读取角度美满Packet数据包

2. //数据包
3. //encode编码：把这种自定义协议对象转换为字节数组
4. //decode解码：拿到一段完整的字节数组之后，解决粘包拆包问题，然后把完整字节数组还原自定义协议对象
5. public class Packet {
6.     private Header header = new Header();
7.     private Body body = new Body();
9.     class Header {
10.         //4个字节检验值：magic number + major version + minor version
11.         private int crc;
12.         //完整header+body大小，解决粘包拆包问题，4个字节大小
13.         private int packetSize;
14.         //节点之间建立的会话的id，物理链路一旦建立好了以后，就必然会分配一个唯一的sessionId，8个字节大小
15.         private long sessionId;
16.         //数据包类型，区分包是个什么，握手请求包，握手应答包
17.         private byte type;
18.         //数据包重要级别
19.         private byte level;
20.         //扩展消息头header字节数量的大小
21.         private int extendHeaderSize;
22.         //扩展消息头
23.         private Map<String, Object> extendHeaders = new HashMap<String, Object>();
24.         //body的字节数量的大小
25.         private int bodySize;
26.     }
28.     class Body {
29.         private byte[] bytes;
30.     }
31. }

复制代码

(3)Packet数据包的编码器实现
一.序列化工具利用Hessian

2. public class SerializeUtils {
3.     public static byte[] serialize(Object object) throws IOException {
4.         ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
5.         HessianOutput hessianOutput = new HessianOutput(byteArrayOutputStream);
6.         hessianOutput.writeObject(object);
7.         byte[] bytes = byteArrayOutputStream.toByteArray();
8.         return bytes;
9.     }
10.    
11.     public static Object deserialize(byte[] bytes, Class clazz) throws IOException {
12.         ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
13.         HessianInput hessianInput = new HessianInput(byteArrayInputStream);
14.         Object object = hessianInput.readObject(clazz);
15.         return object;
16.     }
17. }

复制代码

二.封装Packet数据包时须要举行序列化和字节巨细盘算

2. //Packet数据包编码器
3. public class PacketEncoder extends MessageToMessageEncoder<Packet> {
4.     protected void encode(ChannelHandlerContext ctx, Packet msg, List<Object> out) throws Exception {
5.         Packet packet = msg;
6.         if (packet == null || packet.getHeader() == null) {
7.             throw new ProtocolException("packet or header is null.");
8.         }
9.         //分配一块不在内存池子里的临时内存块，底层就是一个byte[]字节数组
10.         ByteBuf byteBuf = Unpooled.buffer();
11.         byteBuf.writeInt(packet.getHeader().getCrc());
12.         byteBuf.writeInt(packet.getPacketSize());
13.         byteBuf.writeLong(packet.getHeader().getSessionId());
14.         byteBuf.writeByte(packet.getHeader().getType());
15.         byteBuf.writeByte(packet.getHeader().getLevel());
16.         byteBuf.writeInt(packet.getHeader().getExtendHeaders().length);
17.         byteBuf.writeBytes(packet.getHeader().getExtendHeaders());
18.         byteBuf.writeInt(packet.getBody().length);
19.         byteBuf.writeBytes(packet.getBody());
20.         out.add(byteBuf);
21.     }
22. }
24. //数据包
25. //encode编码：把这种自定义协议对象转换为字节数组
26. //decode解码：拿到一段完整的字节数组之后，解决粘包拆包问题，然后把完整字节数组还原自定义协议对象
27. public class Packet {
28.     private static final int ZERO_BYTES = 0;
29.     private static final int SINGLE_BYTES = 1;
30.     private static final int INT_BYTES = 4;
31.     private static final int LONG_BYTES = 8;
32.     private Header header = new Header();
33.     private Object body;
34.     private byte[] bodyBytes;
36.     //完整header+body大小，解决粘包拆包问题，4个字节大小
37.     private int packetSize = ZERO_BYTES;
39.     class Header {
40.         //4个字节检验值：magic number + major version + minor version
41.         private int crc = 0xabef0101;
42.         //节点之间建立的会话的id，物理链路一旦建立好了以后，就必然会分配一个唯一的sessionId，8个字节大小
43.         private long sessionId;
44.         //数据包类型，区分包是个什么，握手请求包，握手应答包
45.         private byte type;
46.         //数据包重要级别
47.         private byte level;
48.         //扩展消息头
49.         private Map<String, Object> extendHeaders = new HashMap<String, Object>();
50.         //扩展消息头序列化后的字节数组
51.         private byte[] extendHeadersBytes;
53.         public int getCrc() { return crc; }
54.         public void setCrc(int crc) { this.crc = crc; }
55.         public long getSessionId() { return sessionId; }
56.         public void setSessionId(long sessionId) {
57.             this.sessionId = sessionId;
58.             packetSize += LONG_BYTES;
59.         }
60.         public byte getType() { return type; }
61.         public void setType(byte type) {
62.             this.type = type;
63.             packetSize += SINGLE_BYTES;
64.         }
65.         public byte getLevel() { return level; }
66.         public void setLevel(byte level) {
67.             this.level = level;
68.             packetSize += SINGLE_BYTES;
69.         }
70.         public byte[] getExtendHeaders() {
71.             return extendHeadersBytes;
72.         }
73.         public void setExtendHeaders(Map<String, Object> extendHeaders) throws IOException {
74.             this.extendHeaders = extendHeaders;
75.             this.extendHeadersBytes = SerializeUtils.serialize(extendHeaders);
76.             packetSize += INT_BYTES;
77.             packetSize += this.extendHeadersBytes.length;
78.         }
79.     }
81.     class Body {
82.         private byte[] bytes;
83.     }
84.    
85.     public int getPacketSize() { return packetSize; }
86.     public void setPacketSize(int packetSize) { this.packetSize = packetSize; }
87.     public Header getHeader() { return header; }
88.     public void setHeader(Header header) { this.header = header; }
89.     public byte[] getBody() { return bodyBytes; }
90.     public void setBody(byte[] body) throws IOException {
91.         this.body = body;
92.         this.bodyBytes = SerializeUtils.serialize(body);
93.         packetSize += INT_BYTES;
94.         packetSize += this.bodyBytes.length;
95.     }
96. }

复制代码

(4)基于builder计划模式重构Packet数据包
Builder计划模式：Builder是一个静态内部类、会new一个Packet实例对象。Builder类中除了build()方法返回实例对象，其他方法都是设置实例对象并返回Builder。此Packet类尚有一个静态方法builder()返回一个静态内部类Builder的实例对象。

2. //数据包
3. //encode编码：把这种自定义协议对象转换为字节数组
4. //decode解码：拿到一段完整的字节数组之后，解决粘包拆包问题，然后把完整字节数组还原自定义协议对象
5. public class Packet {
6.     private static final int ZERO_BYTES = 0;
7.     private static final int SESSION_ID_BYTES = 8;
8.     private static final int TYPE_BYTES = 1;
9.     private static final int LEVEL_BYTES = 1;
10.     private static final int EXTEND_HEADERS_SIZE_BYTES = 4;
11.     private static final int BODY_SIZE_BYTES = 4;
13.     private Header header = new Header();
14.     private Object body = new Body();
15.     private byte[] bodyBytes;
17.     //4个字节检验值：magic number + major version + minor version
18.     private int crc = 0xabef0101;
19.     //完整header+body大小，解决粘包拆包问题，4个字节大小
20.     private int packetSize = ZERO_BYTES;
22.     class Header {
23.         //节点之间建立的会话的id，物理链路一旦建立好了以后，就必然会分配一个唯一的sessionId，8个字节大小
24.         private long sessionId;
25.         //数据包类型，区分包是个什么，握手请求包，握手应答包
26.         private byte type;
27.         //数据包重要级别
28.         private byte level;
29.         //扩展消息头
30.         private Map<String, Object> extendHeaders = new HashMap<String, Object>();
31.         //扩展消息头序列化后的字节数组
32.         private byte[] extendHeadersBytes;
33.         public long getSessionId() { return sessionId; }
34.         public void setSessionId(long sessionId) { this.sessionId = sessionId; }
35.         public byte getType() { return type; }
36.         public void setType(byte type) { this.type = type; }
37.         public byte getLevel() { return level; }
38.         public void setLevel(byte level) { this.level = level; }
39.         public Map<String, Object> getExtendHeaders() { return extendHeaders; }  
40.         public void setExtendHeaders(Map<String, Object> extendHeaders) { this.extendHeaders = extendHeaders; }
41.         public byte[] getExtendHeadersBytes() { return extendHeadersBytes; }
42.         public void setExtendHeadersBytes(byte[] extendHeadersBytes) { this.extendHeadersBytes = extendHeadersBytes; }
43.     }
45.     class Body {
46.         private byte[] bytes;
47.     }
49.     public int getCrc() { return crc; }
50.     public void setCrc(int crc) { this.crc = crc; }
51.     public int getPacketSize() { return packetSize; }
52.     public void setPacketSize(int packetSize) { this.packetSize = packetSize; }
53.     public Header getHeader() { return header; }
54.     public void setHeader(Header header) { this.header = header; }
55.     public Object getBody() { return body; }
56.     public void setBody(Object body) { this.body = body; }
57.     public byte[] getBodyBytes() { return bodyBytes; }
58.     public void setBodyBytes(byte[] bodyBytes) { this.bodyBytes = bodyBytes; }
59.     static class Builder {
60.         private Packet packet = new Packet();
61.         public Builder sessionId(long sessionId) { packet.getHeader().setSessionId(sessionId); return this; }
62.         public Builder type(byte type) { packet.getHeader().setType(type); return this; }
63.         public Builder level(byte level) { packet.getHeader().setLevel(level); return this; }
64.         public Builder extendHeader(String key, Object val) { packet.getHeader().getExtendHeaders().put(key, val); return this; }
65.         public Builder body(Object body) { packet.setBody(body); return this; }
66.         public Packet build() throws IOException {
67.             byte[] extendHeadersBytes = SerializeUtils.serialize(packet.getHeader().getExtendHeaders());
68.             packet.getHeader().setExtendHeadersBytes(extendHeadersBytes);
69.             byte[] bodyBytes = SerializeUtils.serialize(packet.getBody());
70.             packet.setBodyBytes(bodyBytes);
71.             int packageSize = ZERO_BYTES + SESSION_ID_BYTES + TYPE_BYTES + LEVEL_BYTES
72.                 + EXTEND_HEADERS_SIZE_BYTES + packet.getHeader().getExtendHeadersBytes().length
73.                 + BODY_SIZE_BYTES + packet.getBodyBytes().length;
74.             packet.setPacketSize(packageSize);
75.             return packet;
76.         }
77.     }
79.     public static Builder builder() {
80.         return new Builder();
81.     }
82. }
84. //数据包编码器
85. public class PacketEncoder extends MessageToMessageEncoder<Packet> {
86.     protected void encode(ChannelHandlerContext ctx, Packet msg, List<Object> out) throws Exception {
87.         Packet packet = msg;
88.         if (packet == null || packet.getHeader() == null) {
89.             throw new ProtocolException("packet or header is null.");
90.         }
91.         //分配一块不在内存池子里的临时内存块，底层就是一个byte[]字节数组
92.         ByteBuf byteBuf = Unpooled.buffer();
93.         byteBuf.writeInt(packet.getCrc());
94.         byteBuf.writeInt(packet.getPacketSize());
95.         byteBuf.writeLong(packet.getHeader().getSessionId());
96.         byteBuf.writeByte(packet.getHeader().getType());
97.         byteBuf.writeByte(packet.getHeader().getLevel());
98.         byteBuf.writeInt(packet.getHeader().getExtendHeadersBytes().length);
99.         byteBuf.writeBytes(packet.getHeader().getExtendHeadersBytes());
100.         byteBuf.writeInt(packet.getBodyBytes().length);
101.         byteBuf.writeBytes(packet.getBodyBytes());
102.         out.add(byteBuf);
103.     }
104. }

复制代码

(5)Packet数据包的解码器实现
一.解码器解码扩展头字段

2. //数据包解码器
3. public class PacketDecoder extends LengthFieldBasedFrameDecoder {
4.     //解码器会负责粘包和拆包的处理，确保拿到的ByteBuf是一段完整的协议对象字节数据
5.     //frame是一个数据帧，就是一个完整的协议对象字节数据
6.     //maxFrameLength，设定一下这个协议对象的帧数据的最大的大小
7.     //lengthFieldOffset，代表帧数据有多少自己的field，它的offset是哪一位
8.     //从offset那个位置开始读，读lengthFieldLength个字节数量，就可以了
9.     public PacketDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
10.         super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
11.     }
13.     @Override
14.     protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
15.         ByteBuf frame = (ByteBuf) super.decode(ctx, in);//利用父类实现粘包拆包
16.         if (frame == null) {
17.             return null;
18.         }
19.         Packet packet = Packet.builder()
20.             .crc(frame.readInt())
21.             .packetSize(frame.readInt())
22.             .sessionId(frame.readLong())
23.             .type(frame.readByte())
24.             .level(frame.readByte())
25.             .extendHeaders(decodeExtendHeaders(frame))
26.             .body(decodeBody(frame))
27.             .build();
28.         return packet;
29.     }
31.     private Map<String, Object> decodeExtendHeaders(ByteBuf in) throws IOException {
32.         int extendHeadersBytesSize = in.readInt();
33.         byte[] extendHeadersBytes = new byte[extendHeadersBytesSize];
34.         in.readBytes(extendHeadersBytes);
35.         Map<String, Object> extendHeaders = (Map<String, Object>)SerializeUtils.deserialize(extendHeadersBytes, Map.class);
36.         return extendHeaders;
37.     }
38.    
39.     private Object decodeBody(ByteBuf in) throws IOException, ClassNotFoundException {
40.         ...
41.     }
42. }

复制代码

二.解码器解码PacketBody对象
起首给Packet数据包增长bodyClazz字段表明body的类名：

2. //数据包
3. //encode编码：把这种自定义协议对象转换为字节数组
4. //decode解码：拿到一段完整的字节数组之后，解决粘包拆包问题，然后把完整字节数组还原自定义协议对象
5. public class Packet {
6.     private static final int ZERO_BYTES = 0;
7.     private static final int SESSION_ID_BYTES = 8;
8.     private static final int TYPE_BYTES = 1;
9.     private static final int LEVEL_BYTES = 1;
10.     private static final int EXTEND_HEADERS_SIZE_BYTES = 4;
11.     private static final int BODY_SIZE_BYTES = 4;
12.     private static final int BODY_CLAZZ_SIZE_BYTES = 4;
14.     private Header header = new Header();
15.     private Object body = new Body();
16.     private byte[] bodyBytes;
17.     private String bodyClazz;
19.     //4个字节检验值：magic number + major version + minor version
20.     private int crc = 0xabef0101;
21.     //完整header+body大小，解决粘包拆包问题，4个字节大小
22.     private int packetSize = ZERO_BYTES;
24.     class Header {
25.         //节点之间建立的会话的id，物理链路一旦建立好了以后，就必然会分配一个唯一的sessionId，8个字节大小
26.         private long sessionId;
27.         //数据包类型，区分包是个什么，握手请求包，握手应答包
28.         private byte type;
29.         //数据包重要级别
30.         private byte level;
31.         //扩展消息头
32.         private Map<String, Object> extendHeaders = new HashMap<String, Object>();
33.         //扩展消息头序列化后的字节数组
34.         private byte[] extendHeadersBytes;
35.         public long getSessionId() { return sessionId; }
36.         public void setSessionId(long sessionId) { this.sessionId = sessionId; }
37.         public byte getType() { return type; }
38.         public void setType(byte type) { this.type = type; }
39.         public byte getLevel() { return level; }
40.         public void setLevel(byte level) { this.level = level; }
41.         public Map<String, Object> getExtendHeaders() { return extendHeaders; }
42.         public void setExtendHeaders(Map<String, Object> extendHeaders) { this.extendHeaders = extendHeaders; }
43.         public byte[] getExtendHeadersBytes() { return extendHeadersBytes; }
44.         public void setExtendHeadersBytes(byte[] extendHeadersBytes) { this.extendHeadersBytes = extendHeadersBytes; }
45.     }
47.     class Body {
48.         private byte[] bytes;
49.     }
51.     public int getCrc() { return crc; }
52.     public void setCrc(int crc) { this.crc = crc; }
53.     public int getPacketSize() { return packetSize; }
54.     public void setPacketSize(int packetSize) { this.packetSize = packetSize; }
55.     public Header getHeader() { return header; }
56.     public void setHeader(Header header) { this.header = header; }
57.     public Object getBody() { return body; }
58.     public void setBody(Object body) { this.body = body; }
59.     public byte[] getBodyBytes() { return bodyBytes; }
60.     public void setBodyBytes(byte[] bodyBytes) { this.bodyBytes = bodyBytes; }
61.     public String getBodyClazz() { return bodyClazz; }
62.     public void setBodyClazz(String bodyClazz) { this.bodyClazz = bodyClazz; }
63.     static class Builder {
64.         private Packet packet = new Packet();
65.         public Builder crc(int crc) { packet.setCrc(crc); return this; }
66.         public Builder packetSize(int packetSize) { packet.setPacketSize(packetSize); return this; }
67.         public Builder sessionId(long sessionId) { packet.getHeader().setSessionId(sessionId); return this; }
68.         public Builder type(byte type) { packet.getHeader().setType(type); return this; }
69.         public Builder level(byte level) { packet.getHeader().setLevel(level); return this; }
70.         public Builder extendHeader(String key, Object val) { packet.getHeader().getExtendHeaders().put(key, val); return this; }
71.         public Builder extendHeaders(Map<String, Object> extendHeaders) { packet.getHeader().setExtendHeaders(extendHeaders); return this; }
72.         public Builder body(Object body) { packet.setBody(body); return this; }
73.         public Packet build() throws IOException {
74.             if (packet.getBody() == null) {
75.                 throw new ProtocolException("packet body is null.");
76.             }
77.             if (packet.getPacketSize() > 0) {
78.                 return packet;
79.             }
80.             byte[] extendHeadersBytes = SerializeUtils.serialize(packet.getHeader().getExtendHeaders());
81.             packet.getHeader().setExtendHeadersBytes(extendHeadersBytes);
82.             byte[] bodyBytes = SerializeUtils.serialize(packet.getBody());
83.             packet.setBodyBytes(bodyBytes);
84.             packet.setBodyClazz(packet.getBody().getClass().getName());
85.             int packageSize = ZERO_BYTES + SESSION_ID_BYTES + TYPE_BYTES + LEVEL_BYTES
86.                 + EXTEND_HEADERS_SIZE_BYTES + packet.getHeader().getExtendHeadersBytes().length
87.                 + BODY_SIZE_BYTES + packet.getBodyBytes().length
88.                 + BODY_CLAZZ_SIZE_BYTES + packet.getBodyClazz().getBytes().length;
89.             packet.setPacketSize(packageSize);
90.             return packet;
91.         }
92.     }
94.     public static Builder builder() {
95.         return new Builder();
96.     }
97. }

复制代码

下面是对字节数组body部门举行解码的逻辑：

2. //数据包解码器
3. public class PacketDecoder extends LengthFieldBasedFrameDecoder {
4.     //解码器会负责粘包和拆包的处理，确保拿到的ByteBuf是一段完整的协议对象字节数据
5.     //frame是一个数据帧，就是一个完整的协议对象字节数据
6.     //maxFrameLength，设定一下这个协议对象的帧数据的最大的大小
7.     //lengthFieldOffset，代表帧数据有多少自己的field，它的offset是哪一位
8.     //从offset那个位置开始读，读lengthFieldLength个字节数量，就可以了
9.     public PacketDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
10.         super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
11.     }
13.     @Override
14.     protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
15.         ByteBuf frame = (ByteBuf) super.decode(ctx, in);
16.         if (frame == null) {
17.             return null;
18.         }
19.         Packet packet = Packet.builder()
20.             .crc(frame.readInt())
21.             .packetSize(frame.readInt())
22.             .sessionId(frame.readLong())
23.             .type(frame.readByte())
24.             .level(frame.readByte())
25.             .extendHeaders(decodeExtendHeaders(frame))
26.             .body(decodeBody(frame))
27.             .build();
28.         return packet;
29.     }
31.     private Map<String, Object> decodeExtendHeaders(ByteBuf in) throws IOException {
32.         int extendHeadersBytesSize = in.readInt();
33.         byte[] extendHeadersBytes = new byte[extendHeadersBytesSize];
34.         in.readBytes(extendHeadersBytes);
35.         Map<String, Object> extendHeaders = (Map<String, Object>)SerializeUtils.deserialize(extendHeadersBytes, Map.class);
36.         return extendHeaders;
37.     }
39.     private Object decodeBody(ByteBuf in) throws IOException, ClassNotFoundException {
40.         int bodyBytesSize = in.readInt();
41.         byte[] bodyBytes = new byte[bodyBytesSize];
42.         in.readBytes(bodyBytes);
44.         int bodyClazzBytesSize = in.readInt();
45.         byte[] bodyClazzBytes = new byte[bodyClazzBytesSize];
46.         in.readBytes(bodyClazzBytes);
48.         String bodyClazzString = new String(bodyClazzBytes);
49.         Class bodyClazz = Class.forName(bodyClazzString);
50.         Object body = SerializeUtils.deserialize(bodyBytes, bodyClazz);
52.         return body;
53.     }
54. }

复制代码