Netty 协议设计与解析 (自定义协议)
一、自定义协议要素
- 魔术 , 用来在第一时间判断是否是无效数据包
- 版本号，可以支持协议的升级
- 序列化算法， 消息正文到底采用哪种序列化和反序列化方式， 可以由此扩展，例如：
json、protobuf、hessian、jdk(缺点不能跨平台) - 指令类型，是登录，注册，单聊，群聊... 跟业务相关
- 请求需要，为了双工通信，提供异步能力
- 正文长度 ， 通过该长度server ，client 可知接下来要读取多少个字节
- 消息正文 例如：(
json,xml,对象流)
二、自定义协议编解码器
MessageCodec 继承 ByteToMessageCodec
@Slf4j
public class MessageCodec extends ByteToMessageCodec<Message> {
/**
* OutBound 出栈
*/
@Override
protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
// 消息编码
// 1. 魔术 4个字节
out.writeBytes(new byte[]{'s','6','8','6'});
// 2. 版本号 1个字节 版本1
out.writeByte(1);
// 3. 序列化算法 1个字节 约定 0:jdk 1:json
out.writeByte(0);
// 4. 指令类型 1个字节
out.writeByte(msg.getMessageType());
// 5. 请求序号 4个字节
out.writeInt(msg.getSequenceId());
// 添加一个无意义的字节，目的是 将除了消息正文外的 字节数填充为 2的n次方
out.writeByte(0xff);
// 正文 目前使用jdk的序列化方式 (对象流)
// 序列化 msg对象, 并得到 序列化后的对象字节数组
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(msg);
byte[] msgByte = bos.toByteArray();
// 6. 正文长度 4个字节
out.writeInt(msgByte.length);
// 7. 消息正文
out.writeBytes(msgByte);
}
/**
* InBound buf入栈
*/
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
// 1. 魔术 4个字节
int magic = in.readInt();
// 2. 版本号 1个字节
byte version = in.readByte();
// 3. 序列化算法 1个字节
byte serializerType = in.readByte();
// 4. 指令类型 1个字节
byte messageType = in.readByte();
// 5. 请求序号 4个字节
int sequenceId= in.readInt();
// 6. 无意义填充字节
in.readByte();
// 7. 消息正文长度
int length = in.readInt();
// 8. 按照消息正文长度 读取正文
byte[] bytes = new byte[length];
in.readBytes(bytes,0,length);
// 反序列化成Message对象
ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
ObjectInputStream ois = new ObjectInputStream(bis);
Message msg = (Message) ois.readObject();
log.debug("{},{},{},{},{},{}",magic,version,serializerType,messageType,sequenceId,length);
log.debug("{}",msg);
// 给下一个handler去读取
out.add(msg);
}
}
1.出栈测试
public class TestMessageCodec {
public static void main(String[] args) throws Exception {
EmbeddedChannel channel = new EmbeddedChannel(
new LoggingHandler(LogLevel.DEBUG),
new MessageCodec());
LoginRequestMessage reqeustMessage = new LoginRequestMessage("zhangsan", "123456");
// 出栈 encode
channel.writeOutbound(reqeustMessage);
}
}
结果：
00:19:00.364 [main] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] WRITE: 211B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 73 36 38 36 01 00 00 00 00 00 00 ff 00 00 00 c3 |s686............|
|00000010| ac ed 00 05 73 72 00 22 63 6e 2e 7a 7a 70 2e 6d |....sr."cn.zzp.m|
|00000020| 65 73 73 61 67 65 2e 4c 6f 67 69 6e 52 65 71 75 |essage.LoginRequ|
|00000030| 65 73 74 4d 65 73 73 61 67 65 67 00 53 07 0f 85 |estMessageg.S...|
|00000040| 37 73 02 00 02 4c 00 08 70 61 73 73 77 6f 72 64 |7s...L..password|
|00000050| 74 00 12 4c 6a 61 76 61 2f 6c 61 6e 67 2f 53 74 |t..Ljava/lang/St|
|00000060| 72 69 6e 67 3b 4c 00 08 75 73 65 72 6e 61 6d 65 |ring;L..username|
|00000070| 71 00 7e 00 01 78 72 00 16 63 6e 2e 7a 7a 70 2e |q.~..xr..cn.zzp.|
|00000080| 6d 65 73 73 61 67 65 2e 4d 65 73 73 61 67 65 08 |message.Message.|
|00000090| ec 48 b8 06 5c 29 bd 02 00 02 49 00 0b 6d 65 73 |.H..)....I..mes|
|000000a0| 73 61 67 65 54 79 70 65 49 00 0a 73 65 71 75 65 |sageTypeI..seque|
|000000b0| 6e 63 65 49 64 78 70 00 00 00 00 00 00 00 00 74 |nceIdxp........t|
|000000c0| 00 06 31 32 33 34 35 36 74 00 08 7a 68 61 6e 67 |..123456t..zhang|
|000000d0| 73 61 6e |san |
+--------+-------------------------------------------------+----------------+
00:19:00.365 [main] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] FLUSH
2.入栈测试
public class TestMessageCodec {
public static void main(String[] args) throws Exception {
EmbeddedChannel channel = new EmbeddedChannel(
// 添加 LTC节码器 解决 消息的 黏包与半包
new LengthFieldBasedFrameDecoder(1024,12,4,0,0),
new LoggingHandler(LogLevel.DEBUG),
new MessageCodec());
LoginRequestMessage reqeustMessage = new LoginRequestMessage("zhangsan", "123456");
// 入栈 decode
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
new MessageCodec().encode(null,reqeustMessage,buf);
channel.writeInbound(buf);
}
}
入栈结果：
00:22:55.695 [main] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 211B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 73 36 38 36 01 00 00 00 00 00 00 ff 00 00 00 c3 |s686............|
|00000010| ac ed 00 05 73 72 00 22 63 6e 2e 7a 7a 70 2e 6d |....sr."cn.zzp.m|
|00000020| 65 73 73 61 67 65 2e 4c 6f 67 69 6e 52 65 71 75 |essage.LoginRequ|
|00000030| 65 73 74 4d 65 73 73 61 67 65 67 00 53 07 0f 85 |estMessageg.S...|
|00000040| 37 73 02 00 02 4c 00 08 70 61 73 73 77 6f 72 64 |7s...L..password|
|00000050| 74 00 12 4c 6a 61 76 61 2f 6c 61 6e 67 2f 53 74 |t..Ljava/lang/St|
|00000060| 72 69 6e 67 3b 4c 00 08 75 73 65 72 6e 61 6d 65 |ring;L..username|
|00000070| 71 00 7e 00 01 78 72 00 16 63 6e 2e 7a 7a 70 2e |q.~..xr..cn.zzp.|
|00000080| 6d 65 73 73 61 67 65 2e 4d 65 73 73 61 67 65 08 |message.Message.|
|00000090| ec 48 b8 06 5c 29 bd 02 00 02 49 00 0b 6d 65 73 |.H..)....I..mes|
|000000a0| 73 61 67 65 54 79 70 65 49 00 0a 73 65 71 75 65 |sageTypeI..seque|
|000000b0| 6e 63 65 49 64 78 70 00 00 00 00 00 00 00 00 74 |nceIdxp........t|
|000000c0| 00 06 31 32 33 34 35 36 74 00 08 7a 68 61 6e 67 |..123456t..zhang|
|000000d0| 73 61 6e |san |
+--------+-------------------------------------------------+----------------+
00:22:55.697 [main] DEBUG cn.zzp.protocol.MessageCodec - 1932933174,1,0,0,0,195
00:22:55.697 [main] DEBUG cn.zzp.protocol.MessageCodec - LoginRequestMessage(super=Message(sequenceId=0, messageType=0), username=zhangsan, password=123456)
00:22:55.697 [main] DEBUG io.netty.handler.logging.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ COMPLETE
从结果可得到消息：
LoginRequestMessage(super=Message(sequenceId=0, messageType=0), username=zhangsan, password=123456)