自定义兼容多种Protobuf协议的编解码器

时间：2022-01-07 06:19:05

《从零开始搭建游戏服务器》自定义兼容多种Protobuf协议的编解码器

直接在protobuf序列化数据的前面，加上一个自定义的协议头，协议头里包含序列数据的长度和对应的数据类型，在数据解包的时候根据包头来进行反序列化。

1.协议头定义

关于这一块，我打算先采取比较简单的办法，结构如下： 
自定义兼容多种Protobuf协议的编解码器 
协议号是自定义的一个int类型的枚举（当然，假如协议吧比较少的话，可以用一个short来代替int以缩小数据包），这个协议号与协议类型是一一对应的，而协议头通常使用数据总长度来填入，具体过程如下：

  • 当客户端向服务器发送数据时，会根据协议类型加上协议号，然后使用protobuf序列化之后再发送给服务器；
  • 当服务器发送数据给客户端时，根据协议号，确定protobuf协议类型以反序列化数据，并调用相应回调方法。

2.自定义的编码器和解码器

编码器： 
参考netty自带的编码器ProtobufEncoder可以发现，被绑定到ChannelPipeline上用于序列化协议数据的编码器，必须继承MessageToByteEncoder<MessageLite>这个基类，并通过重写protected void encode(ChannelHandlerContext ctx, MessageLite msg, ByteBuf out)这个方法来实现自定义协议格式的目的：

package com.tw.login.tools;

import com.google.protobuf.MessageLite;

import com.tw.login.proto.CsEnum.EnmCmdID;

import com.tw.login.proto.CsLogin;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.MessageToByteEncoder;

/**

 * 自定义编码器

 * @author linsh

 *

 */

public class PackEncoder extends MessageToByteEncoder<MessageLite> {

    /**

     * 传入协议数据，产生携带包头之后的数据

     */

    @Override

    protected void encode(ChannelHandlerContext ctx, MessageLite msg, ByteBuf out) throws Exception {

        // TODO Auto-generated method stub

        byte[] body = msg.toByteArray();

        byte[] header = encodeHeader(msg, (short)body.length);

        out.writeBytes(header);

        out.writeBytes(body);

        return;

    }

    /**

     * 获得一个协议头

     * @param msg

     * @param bodyLength

     * @return

     */

    private byte[] encodeHeader(MessageLite msg,short bodyLength){

        short _typeId = 0;

        if(msg instanceof CsLogin.CSLoginReq){

            _typeId = EnmCmdID.CS_LOGIN_REQ_VALUE;

        }else if(msg instanceof CsLogin.CSLoginRes){

            _typeId = EnmCmdID.CS_LOGIN_RES_VALUE;

        }

        //存放两个short数据

        byte[] header = new byte[4];

        //前两位放数据长度

        header[0] = (byte) (bodyLength & 0xff);

        header[1] = (byte) ((bodyLength >> 8) & 0xff);

        //后两个字段存协议id

        header[2] = (byte) (_typeId & 0xff);

        header[3] = (byte) ((_typeId >> 8) & 0xff);

        return header;

    }

}

解码器： 
参考netty自带的编码器ProtobufDecoder可以发现，被绑定到ChannelPipeline上用于序列化协议数据的解码器，必须继承ByteToMessageDecoder这个基类，并通过重写protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)这个方法来实现解析自定义协议格式的目的：

package com.tw.login.tools;

import java.util.List;

import com.google.protobuf.MessageLite;

import com.tw.login.proto.CsEnum.EnmCmdID;

import com.tw.login.proto.CsLogin.CSLoginReq;

import com.tw.login.proto.CsLogin.CSLoginRes;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.ByteToMessageDecoder;

public class PackDecoder extends ByteToMessageDecoder {

    @Override

    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

        // 获取包头中的body长度

        byte low = in.readByte();

        byte high = in.readByte();

        short s0 = (short) (low & 0xff);

        short s1 = (short) (high & 0xff);

        s1 <<= 8;

        short length = (short) (s0 | s1);

        // 获取包头中的protobuf类型

        byte low_type = in.readByte();

        byte high_type = in.readByte();

        short s0_type = (short) (low_type & 0xff);

        short s1_type = (short) (high_type & 0xff);

        s1_type <<= 8;

        short dataTypeId = (short) (s0_type | s1_type);

        // 如果可读长度小于body长度，恢复读指针，退出。

        if (in.readableBytes() < length) {

            in.resetReaderIndex();

            return;

        }

        //开始读取核心protobuf数据

        ByteBuf bodyByteBuf = in.readBytes(length);

        byte[] array;

        //反序列化数据的起始点

        int offset;

        //可读的数据字节长度

        int readableLen= bodyByteBuf.readableBytes();

        //分为包含数组数据和不包含数组数据两种形式

        if (bodyByteBuf.hasArray()) {

            array = bodyByteBuf.array();

            offset = bodyByteBuf.arrayOffset()   bodyByteBuf.readerIndex();

        } else {

            array = new byte[readableLen];

            bodyByteBuf.getBytes(bodyByteBuf.readerIndex(), array, 0, readableLen);

            offset = 0;

        }

        //反序列化

        MessageLite result = decodeBody(dataTypeId, array, offset, readableLen);

        out.add(result);

    }

    /**

     * 根据协议号用响应的protobuf类型来解析协议数据

     * @param _typeId

     * @param array

     * @param offset

     * @param length

     * @return

     * @throws Exception

     */

    public MessageLite decodeBody(int _typeId,byte[] array,int offset,int length) throws Exception{

        if(_typeId == EnmCmdID.CS_LOGIN_REQ_VALUE){

            return CSLoginReq.getDefaultInstance().getParserForType().parseFrom(array,offset,length);

        }

        else if(_typeId == EnmCmdID.CS_LOGIN_RES_VALUE){

            return CSLoginRes.getDefaultInstance().getParserForType().parseFrom(array,offset,length);

        }

        return null;

    }

}

3.修改Socket管道绑定的编解码器：

在创建Socket管道的时候，将编解码器替换为自定义的编解码器，而具体数据发送和接受过程无需做任何修改：

ChannelPipeline pipeline = ch.pipeline();

// 协议数据的编解码器

pipeline.addLast("frameDecoder",new ProtobufVarint32FrameDecoder());

pipeline.addLast("protobufDecoder",new PackDecoder());

pipeline.addLast("frameEncoder",new ProtobufVarint32LengthFieldPrepender());

pipeline.addLast("protobufEncoder", new PackEncoder());

pipeline.addLast("handler",new SocketServerHandler());

相关文章