自定义兼容多种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());