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)