unity3d+网络模块：protobuf，协议包组成，拆包黏包，多协程接收，网络协议派发，大端小端，压缩，加密

立羽

发布于 2023-08-24 15:18:29

3200

发布于 2023-08-24 15:18:29

protobuf转字节流

[ProtoContract]
public class TestProto
{
    [ProtoMember(1)]
    public long accountId;
    [ProtoMember(2)]
    public string password;
}

        /// <summary>
        /// 序列化pb数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="t"></param>
        /// <returns></returns>
        public static byte[] NSerialize<T>(T t)
        {
            byte[] buffer = null;

            using (MemoryStream m = new MemoryStream())
            {
                Serializer.Serialize<T>(m, t);

                m.Position = 0;
                int length = (int)m.Length;
                buffer = new byte[length];
                m.Read(buffer, 0, length);
            }

            return buffer;
        }

协议包组成

由包头信息，内容字节流流组成，内容直接流即protobuf转字节流其中，包头

//包头信息
    public class ProtocolHead
    {
        public int packetLength  = 0; //整个包的长度：长度字节4个  + modelid字节2个 +cmd字节2个 +内容长度
        public short moduleId = 0; //和cmd组成一条协议的id
        public short cmd = 0;

序列化一条协议

先序列化头

public NetBuffer Serialize(NetBuffer buffer)
        {
            buffer.WriteInt(packetLength);
            buffer.WriteShort(moduleId);
            
            buffer.WriteShort(cmd);
            return buffer;
        }

其中序列化int ，short，发送的是大端模式

public int WriteShort(short value, int writePos = -1)
        {
            int pos = UpdateLenAndGetWritePos(writePos, 2);
            m_buff[pos + 0] = (byte)(value >> 8 & 0xFF);
            m_buff[pos + 1] = (byte)(value >> 0 & 0xFF);
            return pos + 2;
        }
        public int WriteInt(int value, int writePos = -1)
        {
            int pos = UpdateLenAndGetWritePos(writePos, 4);
            m_buff[pos + 0] = (byte)(value >> 24 & 0xFF);
            m_buff[pos + 1] = (byte)(value >> 16 & 0xFF);
            m_buff[pos + 2] = (byte)(value >> 08 & 0xFF);
            m_buff[pos + 3] = (byte)(value >> 00 & 0xFF);
            return pos + 4;
        }

int占4个直接，用m_buff字节数组里4位表示，按照高位在前，低位在后顺序再把内容字节流copy进入m_buff

public int WriteBytes(byte[] src, int srcOffset, int count, int writePos = -1)
        {
            int pos = UpdateLenAndGetWritePos(writePos, count);
            Buffer.BlockCopy(src, srcOffset, m_buff, pos, count);
            return pos + count;
        }

网络发送

byte[] m_sendBuf = new byte[4096];
        public void SendMsg(NetMessage netMsg, EnSocket type = EnSocket.Game)
        {
            //byte[] tmp = null;
            int len = netMsg.Serialize(out m_sendBuf);
            //byte[] buf1 = new byte[len];
            //Array.Copy(tmp, buf1, len);

            clientSocket.BeginSend(m_sendBuf, 0, len, SocketFlags.None, new AsyncCallback(_onSendMsg), clientSocket);
        }

避免反复new字节流产生GC，使用m_sendBuf缓存

多线程接收

private void _onConnect_Sucess(IAsyncResult iar)
        {
            try
            {
                Socket client = (Socket)iar.AsyncState;
                client.EndConnect(iar);

                receiveThread = new Thread(new ThreadStart(_onReceiveSocket));
                receiveThread.IsBackground = true;
                receiveThread.Start();
                _isConnected = true;
                
                m_isContecting = false;

拆包黏包处理

主要思想：

网络接收到数据，往待处理字节流数组a保存；
多了，a会扩容；
每次处理完一条完整协议b，a截取掉前面b所有的字节流数据后，尾部的未处理字节流又组成新的_buff即a 多线程频繁调用，可以Thread.Sleep(100);进行定时获取缓冲中网络数据

int receiveLength = clientSocket.Receive(_tmpReceiveBuff); //每次只要有数据来了，就写入到_tmpReceiveBuff中，返回接收到的长度
if (receiveLength > 0)
{
    _databuffer.AddBuffer(_tmpReceiveBuff, receiveLength);//将收到的数据添加到缓存器中
    while (_databuffer.GetData(out _socketData))//取出一条完整数据
    {
        sEvent_NetMessageData tmpNetMessageData = new sEvent_NetMessageData();
        tmpNetMessageData._eventType = _socketData._protocallType;
        tmpNetMessageData._eventData = _socketData._data;
        tmpNetMessageData.m_key = _socketData.key;
        //锁死消息中心消息队列，并添加数据
        lock (MessageCenter.Instance._netMessageDataQueue)
        {
            //Debug.Log("Get Server:" + tmpNetMessageData.m_key);
            MessageCenter.Instance._netMessageDataQueue.Enqueue(tmpNetMessageData);
        }
    }
}
          }

每次数据来了，塞到包缓冲器里这里包缓冲字节流： private byte[] _buff; //待处理字节流:网络接收到数据，往这里塞；多了，会扩容；每次处理完一条完整协议a，截取掉前面a所有的数据后，尾部的未处理直接流又组成新的_buff

/// <summary>
    /// 添加缓存数据
    /// </summary>
    /// <param name="_data"></param>
    /// <param name="_dataLen"></param>
    public void AddBuffer(byte[] _data, int _dataLen)
    {
        if (_dataLen > _buff.Length - _curBuffPosition)//接收的长度，要塞入_buff中，_buff剩余容量不够，扩容
        {
            byte[] _tmpBuff = new byte[_curBuffPosition + _dataLen];
            Array.Copy(_buff, 0, _tmpBuff, 0, _curBuffPosition);
            Array.Copy(_data, 0, _tmpBuff, _curBuffPosition, _dataLen);
            _buff = _tmpBuff; //生成新的扩容后_buff
            _tmpBuff = null;
        }
        else //剩余空间还够，直接塞入
        {
            Array.Copy(_data, 0, _buff, _curBuffPosition, _dataLen);
        }
        _curBuffPosition += _dataLen;//修改当前数据标记
    }

如果包缓冲器能完整取出一条协议进行处理

/// <summary>
    /// 获取一条可用数据，返回值标记是否有数据
    /// </summary>
    /// <param name="_tmpSocketData"></param>
    /// <returns></returns>
    public bool GetData(out sSocketData _tmpSocketData)
    {
        _tmpSocketData = new sSocketData();

        //_buffLength如果没提取过为 0 ，提取一次，取全包长（4+2+2+内容字节流），使用后又重置为 0 
        if (_buffLength <= 0)
        {
            UpdateDataLength();
        }

        if (_buffLength > 0 && _curBuffPosition >= _buffLength)
        {
            _tmpSocketData._buffLength = _buffLength;
            _tmpSocketData._dataLength = _dataLength;
            _tmpSocketData._protocallType = (eProtocalCommand)_protocalType;
            _tmpSocketData.key = m_key;
            _tmpSocketData._data = new byte[_dataLength];
            Array.Copy(_buff, Constants.HEAD_LEN, _tmpSocketData._data, 0, _dataLength); //_buff 中从 （4+2+2）开始，复制给内容字节流
            _curBuffPosition -= _buffLength; //当前接收到一条网络数据流里还未处理完的字节流 长度 =  总长度（当前长度） - _buffLength（一条完整数据长度）
            byte[] _tmpBuff = new byte[_curBuffPosition < _minBuffLen ? _minBuffLen : _curBuffPosition];
            Array.Copy(_buff, _buffLength, _tmpBuff, 0, _curBuffPosition);
            _buff = _tmpBuff; //重新复制新的待处理字节流


            _buffLength = 0;
            _dataLength = 0;
            _protocalType = 0;
            return true;
        }
        return false;
    }

从缓冲字节流里解析出一条完整协议

/// <summary>
    /// 更新数据长度
    /// </summary>
    public void UpdateDataLength()
    {
        if (_dataLength == 0 && _curBuffPosition >= Constants.HEAD_LEN)
        {
            //从0号位提取4位包长字节流
            byte[] tmpDataLen = new byte[Constants.HEAD_DATA_LEN];
            Array.Copy(_buff, 0, tmpDataLen, 0, Constants.HEAD_DATA_LEN);
            //小端接收，要转换下，转换位包长int
            _buffLength = BitConverter.ToInt32(NetBuffer.ReverseOrder(tmpDataLen), 0)+4; //得到包长度

            //提取moudleID
            byte[] tmpProtocalType = new byte[Constants.HEAD_TYPE_LEN];
            Array.Copy(_buff, Constants.HEAD_DATA_LEN, tmpProtocalType, 0, Constants.HEAD_TYPE_LEN);
            ushort module = BitConverter.ToUInt16(NetBuffer.ReverseOrder(tmpProtocalType), 0);

            //提取cmdID
            byte[] tmpCmd = new byte[Constants.HEAD_TYPE_LEN];
            Array.Copy(_buff, Constants.HEAD_DATA_LEN + Constants.HEAD_TYPE_LEN, tmpCmd, 0, Constants.HEAD_TYPE_LEN);
            ushort cmd = BitConverter.ToUInt16(NetBuffer.ReverseOrder(tmpCmd), 0);

            m_key = module.ToString() + "," + cmd.ToString();

            //内容字节流为全长度 - （4+2+2）
            _dataLength = _buffLength - Constants.HEAD_LEN;
        }
    }

网络协议派发

主要功能： 1.消息放入队列 2.协议底层解析好数据，通过委托，被多个object调取网络接收到一条完整消息，放入到消息队列中

//锁死消息中心消息队列，并添加数据
                            lock (MessageCenter.Instance._netMessageDataQueue)
                            {
                                //Debug.Log("Get Server:" + tmpNetMessageData.m_key);
                                MessageCenter.Instance._netMessageDataQueue.Enqueue(tmpNetMessageData);
                            }

在mono的fixupdate中按照先进先出原则派发消息

while (_netMessageDataQueue.Count > 0)
        {
            lock (_netMessageDataQueue)
            {
                sEvent_NetMessageData tmpNetMessageData = _netMessageDataQueue.Dequeue();
                                if (tmpNetMessageData.m_key != MsgIdDefine.RspPlayerSync)
                {
                    Debug.Log("Get Server:" + tmpNetMessageData.m_key);
                }
                NetEventMgr.Instance.DispatchEvent(tmpNetMessageData.m_key, tmpNetMessageData._eventData);

            }
        }

监听者注册消息，同时把该消息id对应的解析类型注册进入，如果多个object注册同个msgID，进行委托合并Delegate.Combine

class ListenerHelper
{
    public Type TMsg = null;
    public Delegate onMsg;
}

public void AddListener<TMsg>(string cmd, Action<TMsg> onMsg)
    {
        if (m_dicMsgListener.ContainsKey(cmd) == false)
        {
            ListenerHelper helper = new ListenerHelper()
            {
                TMsg = typeof(TMsg),
                onMsg = onMsg
            };

            m_dicMsgListener.Add(cmd, helper);

        }
        else
        {
            m_dicMsgListener[cmd].onMsg = Delegate.Combine(m_dicMsgListener[cmd].onMsg, onMsg);

        }
    }

派发消息，在底层解析出数据，可以通过打网络log，方便查看具体数据

public void DispatchEvent(string cmd, byte[] buf)
    {
        try
        {
            if (m_dicMsgListener.ContainsKey(cmd))
            {
                var helper = m_dicMsgListener[cmd];
                if (helper != null)
                {
                    if (helper.TMsg != null)
                    {
                        object obj = PBSerializer.NDeserialize(buf, helper.TMsg);
                        if (obj != null)
                        {
                            if (DataMgr.m_isNetLog == true)
                            {
                                string log = JsonConvert.SerializeObject(obj);
                                if (cmd != MsgIdDefine.RspPlayerSync /*&& cmd != MsgIdDefine.RspMechanism*/)
                                {
                                    Debug.Log("NetRecv-->Key:" + cmd + "-->" + log);
                                }
                            }

                            helper.onMsg.DynamicInvoke(obj);
                        }
                    }
                    else
                    {
                        if (DataMgr.m_isNetLog == true)
                        {
                            if (cmd != MsgIdDefine.RspPlayerSync/* && cmd != MsgIdDefine.RspMechanism*/)
                            {
                                Debug.Log("NetRecv-->Key:" + cmd);
                            }
                        }
                        helper.onMsg.DynamicInvoke();
                    }
                }
            }
        }
        catch (Exception e)
        {
            Debug.Log("DispatchEvent:(" + cmd + ")--" + e);
        }
    }

大端小端模式

C#大端模式和小端模式。小端（little-endian）模式：低地址上存放低字节，高地址上存放高字节。如0x11223344→ byte[] numBytes = new byte[]{ 0x44,0x33,0x22,0x11}; numBytes[0] = 0x44; //低地址存放低字节 numBytes[3] = 0x11; //高地址存放高字节反之，高字节在前，低字节在后，则为大端模式。反转示例： short num = 12; byte[] bytes = BitConverter.GetBytes(s); Array.Reverse(bytes); //bytes转换为倒序（反转），可实现大端小端的转换大端模式下int转字节流

 m_buff[pos + 0] = (byte)(value >> 24 & 0xFF);
            m_buff[pos + 1] = (byte)(value >> 16 & 0xFF);
            m_buff[pos + 2] = (byte)(value >> 08 & 0xFF);
            m_buff[pos + 3] = (byte)(value >> 00 & 0xFF);

确定服务器采用的是大小端模式，在客户端收发时进行大端小端处理

字节流压缩

使用GZip

 public static byte[] Compress(byte[] binary)
    {
        MemoryStream ms = new MemoryStream();
        GZipOutputStream gzip = new GZipOutputStream(ms);
        //gzip.SetLevel(-1);
        //Debug.Log("gzip.GetLevel()" + gzip.GetLevel());
        gzip.Write(binary, 0, binary.Length);
        gzip.Close();
        byte[] press = ms.ToArray();
        return press;
    }

    public static byte[] DeCompress(byte[] press)
    {
        GZipInputStream gzi = new GZipInputStream(new MemoryStream(press));
        MemoryStream re = new MemoryStream();
        int count = 0;
        int len = press.Length;
        byte[] data = new byte[len];
        while ((count = gzi.Read(data, 0, data.Length)) != 0)
        {
            re.Write(data, 0, count);
        }
        byte[] depress = re.ToArray();
        return depress;
    }