C# Socket系列三 socket通信的封包和拆包

通过系列二 我们已经实现了socket的简单通信 接下来我们测试一下，在时间应用的场景下，我们会快速且大量的传输数据的情况！

   class Program
    {
        static void Main(string[] args)
        {
            TCPListener tcp = new TCPListener();
            TSocketClient client = new TSocketClient();
            for (int i = 0; i < 10; i++)
            {
                client.SendMsg(System.Text.UTF8Encoding.Default.GetBytes("Holle Server！"));
            }
            Console.ReadLine();
        }
    }

我们通过测试代码快速发送10条消息到服务器去，

我们看看运行结果

这样不难看出，我们的客户端发送了10条消息，但是服务器收到的时候变成了两条消息，回复客户端自然就变成两次回复。

这是为什么呢？

我们修改一下程序一秒钟发送一次消息试试

   class Program
    {
        static void Main(string[] args)
        {
            TCPListener tcp = new TCPListener();
            TSocketClient client = new TSocketClient();
            for (int i = 0; i < 5; i++)
            {
                Thread.Sleep(1000);
                client.SendMsg(System.Text.UTF8Encoding.Default.GetBytes("Holle Server！"));
            }
            Console.ReadLine();
        }
    }

运行看看，

这次对了那么分析分析到底为什么呢？
这是socket的底层，做的手脚。因为我设置socket的发送和接受缓冲区
//10K的缓冲区空间 private int BufferSize = 10 * 1024; 10k的缓冲区，
且socket的底层 发送消息会有一定间隙，虽然这个时间很短，但是我们直接for循环发送的话，时间同意很快，
因为socket.send()方法并非真实的发送数据而是把数据压入发送缓冲区。
那么我们就明白了为什么会出现上面的情况
出现了这样的情况我们要怎么解决呢？
时间应用场景不可能1秒钟才一条消息啥。
我们知道了导致这个问题的原因是因为消息发送是出现了快速压入很多发送消息到待发送缓冲区里面一起发送导致的。这样情况就是粘包了，那么我们是不是可以考虑给每一个消息加入包标识呢？

接下来我们修改一下发送包的数据代码

创建消息的构造体 TSocketMessage

 /// <summary>
    /// 底层通信消息
    /// </summary>
    public class TSocketMessage : IDisposable
    {
        /// <summary>
        /// 消息ID
        /// </summary>
        public int MsgID;
        /// <summary>
        /// 消息内容
        /// </summary>
        public byte[] MsgBuffer;

        public TSocketMessage(int msgID, byte[] msg)
        {
            this.MsgID = msgID;
            this.MsgBuffer = msg;
        }

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool flag1)
        {
            if (flag1) { this.MsgBuffer = null; }
        }
    }

接下来我们创建消息包的封装和拆分 MarshalEndian

 public class MarshalEndian
    {
        //用于存储剩余未解析的字节数
        private List<byte> _LBuff = new List<byte>(2);
        //默认是utf8的编码格式
        private UTF8Encoding utf8 = new UTF8Encoding();

        //包头1
        const Int16 t1 = 0x55;
        //包头2
        const Int16 t2 = 0xAA;
        //字节数常量 两个包头4个字节，一个消息id4个字节，封装消息长度 long 8个字节
        const long ConstLenght = 12L;

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool flag1)
        {
            if (flag1)
            {
                IDisposable disposable2 = this.utf8 as IDisposable;
                if (disposable2 != null) { disposable2.Dispose(); }
                IDisposable disposable = this._LBuff as IDisposable;
                if (disposable != null) { disposable.Dispose(); }
            }
        }

        public byte[] Encode(TSocketMessage msg)
        {
            MemoryStream ms = new MemoryStream();
            BinaryWriter bw = new BinaryWriter(ms, new UTF8Encoding());
            byte[] msgBuffer = msg.MsgBuffer;

            #region 封装包头
            bw.Write((Int16)t1);
            bw.Write((Int16)t2);
            #endregion

            #region 包协议
            if (msgBuffer != null)
            {
                bw.Write((Int64)(msgBuffer.Length + 4));
                bw.Write(msg.MsgID);
                bw.Write(msgBuffer);
            }
            else { bw.Write((Int64)0); }
            #endregion

            bw.Close();
            ms.Close();
            bw.Dispose();
            ms.Dispose();
            return ms.ToArray();
        }

        public List<TSocketMessage> GetDcAppMess(byte[] buff, int len)
        {
            //拷贝本次的有效字节
            byte[] _b = new byte[len];
            Array.Copy(buff, 0, _b, 0, _b.Length);
            buff = _b;
            if (this._LBuff.Count > 0)
            {
                //拷贝之前遗留的字节
                this._LBuff.AddRange(_b);
                buff = this._LBuff.ToArray();
                this._LBuff.Clear();
                this._LBuff = new List<byte>(2);
            }

            List<TSocketMessage> list = new List<TSocketMessage>();
            MemoryStream ms = new MemoryStream(buff);
            BinaryReader buffers = new BinaryReader(ms, this.utf8);
            try
            {
                byte[] _buff;
            Label_0073:
                //判断本次解析的字节是否满足常量字节数 
                if ((buffers.BaseStream.Length - buffers.BaseStream.Position) < ConstLenght)
                {
                    _buff = new byte[(int)(buffers.BaseStream.Length - buffers.BaseStream.Position)];
                    Array.Copy(buff, (int)buffers.BaseStream.Position, _buff, 0, _buff.Length);
                    this._LBuff.AddRange(_buff);
                    return list;
                }
                #region 包头读取
            //循环读取包头
            Label_00983:
                Int16 tt1 = buffers.ReadInt16();
                Int16 tt2 = buffers.ReadInt16();
                if (!(tt1 == t1 && tt2 == t2))
                {
                    long ttttt = buffers.BaseStream.Seek(-3, SeekOrigin.Current);
                    goto Label_00983;
                }
                #endregion

                #region 包协议
                long offset = buffers.ReadInt64();
                #endregion

                #region 包解析
                //剩余字节数大于本次需要读取的字节数
                if (offset < (buffers.BaseStream.Length - buffers.BaseStream.Position))
                {
                    int msgID = buffers.ReadInt32();
                    _buff = new byte[offset - 4];
                    Array.Copy(buff, (int)buffers.BaseStream.Position, _buff, 0, _buff.Length);
                    list.Add(new TSocketMessage(msgID, _buff));
                    //设置偏移量 然后继续循环读取
                    buffers.BaseStream.Seek(offset, SeekOrigin.Current);
                    goto Label_0073;
                }
                else if (offset == (buffers.BaseStream.Length - buffers.BaseStream.Position))
                {
                    int msgID = buffers.ReadInt32();
                    //剩余字节数刚好等于本次读取的字节数
                    _buff = new byte[offset - 4];
                    Array.Copy(buff, (int)buffers.BaseStream.Position, _buff, 0, _buff.Length);
                    list.Add(new TSocketMessage(msgID, _buff));
                }
                else
                {
                    //剩余字节数刚好小于本次读取的字节数 存起来，等待接受剩余字节数一起解析
                    _buff = new byte[(int)(buffers.BaseStream.Length - buffers.BaseStream.Position + ConstLenght)];
                    Array.Copy(buff, (int)(buffers.BaseStream.Position - ConstLenght), _buff, 0, _buff.Length);
                    buff = _buff;
                    this._LBuff.AddRange(_buff);
                }
                #endregion

            }
            catch { }
            finally
            {
                if (buffers != null) { buffers.Dispose(); }
                buffers.Close();
                if (buffers != null) { buffers.Dispose(); }
                ms.Close();
                if (ms != null) { ms.Dispose(); }
            }
            return list;
        }
    }

接下来我们修改一下 TSocketBase 的 抽象方法  

 public abstract void Receive(TSocketMessage msg);

在修改接受消息回调函数

  /// <summary>
        /// 消息解析器
        /// </summary>
        MarshalEndian mersha = new MarshalEndian();

        /// <summary>
        /// 接收消息回调函数
        /// </summary>
        /// <param name="iar"></param>
        private void ReceiveCallback(IAsyncResult iar)
        {
            if (!this.IsDispose)
            {
                try
                {
                    //接受消息
                    ReceiveSize = _Socket.EndReceive(iar, out ReceiveError);
                    //检查状态码
                    if (!CheckSocketError(ReceiveError) && SocketError.Success == ReceiveError)
                    {
                        //判断接受的字节数
                        if (ReceiveSize > 0)
                        {
                            byte[] rbuff = new byte[ReceiveSize];
                            Array.Copy(this.Buffers, rbuff, ReceiveSize);
                            var msgs = mersha.GetDcAppMess(rbuff, ReceiveSize);
                            foreach (var msg in msgs)
                            {
                                this.Receive(msg);
                            }
                            //重置连续收到空字节数
                            ZeroCount = 0;
                            //继续开始异步接受消息
                            ReceiveAsync();
                        }
                        else
                        {
                            ZeroCount++;
                            if (ZeroCount == 5) { this.Close("错误链接"); }
                        }
                    }
                }
                catch (System.Net.Sockets.SocketException) { this.Close("链接已经被关闭"); }
                catch (System.ObjectDisposedException) { this.Close("链接已经被关闭"); }
            }
        }

这样我们完成了在收到消息后对数据包的解析。

修改一下TSocketClient的 Receive 重写方法

        /// <summary>
        /// 收到消息后
        /// </summary>
        /// <param name="rbuff"></param>
        public override void Receive(TSocketMessage msg)
        {
            Console.WriteLine("Receive ID：" + msg.MsgID + " Msg：" + System.Text.UTF8Encoding.Default.GetString(msg.MsgBuffer));
            if (isServer)
            {
                this.SendMsg(new TSocketMessage(msg.MsgID, System.Text.UTF8Encoding.Default.GetBytes("Holle Client！")));
            }
        }

修改测试代码如下

    class Program
    {
        static void Main(string[] args)
        {
            TCPListener tcp = new TCPListener();
            TSocketClient client = new TSocketClient();
            for (int i = 1; i < 5; i++)
            {
                Thread.Sleep(1000);
                client.SendMsg(new TSocketMessage(i, System.Text.UTF8Encoding.Default.GetBytes("Holle Server！")));
            }
            Console.ReadLine();
        }
    }

运行结果

接受成功了，那么我们取消暂停状态，快速发送消息试试

    class Program
    {
        static void Main(string[] args)
        {
            TCPListener tcp = new TCPListener();
            TSocketClient client = new TSocketClient();
            for (int i = 1; i < 5; i++)
            {
                client.SendMsg(new TSocketMessage(i, System.Text.UTF8Encoding.Default.GetBytes("Holle Server！")));
            }
            Console.ReadLine();
        }
    }

看看运行结果

瞬间完成了消息发送，也没有再出现第一次运行的那样~！

这样完美的解决了socket通信 在传输上发送粘包问题

下载程序完整代码

谢谢园友发现的问题，

问题是这样的原本的解包和封包的测试代码不够严谨导致解析包出现错误
感谢园友发现问题，并提出问题。

附上最新的代码

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace TSocket
{
    public class MarshalEndian
    {
        //用于存储剩余未解析的字节数
        private List<byte> _LBuff = new List<byte>(2);
        //默认是utf8的编码格式
        private UTF8Encoding utf8 = new UTF8Encoding();

        //包头1
        const Int16 t1 = 0x55;
        //包头2
        const Int16 t2 = 0xAA;
        //字节数常量 两个包头4个字节，一个消息id4个字节，封装消息长度 int32 4个字节
        const Int32 ConstLenght = 8;

        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool flag1)
        {
            if (flag1)
            {
                IDisposable disposable2 = this.utf8 as IDisposable;
                if (disposable2 != null) { disposable2.Dispose(); }
                IDisposable disposable = this._LBuff as IDisposable;
                if (disposable != null) { disposable.Dispose(); }
            }
        }

        public byte[] Encode(TSocketMessage msg)
        {
            MemoryStream ms = new MemoryStream();
            BinaryWriter bw = new BinaryWriter(ms, new UTF8Encoding());
            byte[] msgBuffer = msg.MsgBuffer;

            #region 封装包头
            bw.Write((Int16)t1);
            bw.Write((Int16)t2);
            #endregion

            #region 包协议
            if (msgBuffer != null)
            {
                bw.Write((Int32)(msgBuffer.Length + 4));
                bw.Write(msg.MsgID);
                bw.Write(msgBuffer);
            }
            else { bw.Write((Int32)0); }
            #endregion

            bw.Close();
            ms.Close();
            bw.Dispose();
            ms.Dispose();
            return ms.ToArray();
        }

        public List<TSocketMessage> GetDcAppMess(byte[] buff, int len)
        {
            //拷贝本次的有效字节
            byte[] _b = new byte[len];
            Array.Copy(buff, 0, _b, 0, _b.Length);
            buff = _b;
            if (this._LBuff.Count > 0)
            {
                //拷贝之前遗留的字节
                this._LBuff.AddRange(_b);
                buff = this._LBuff.ToArray();
                this._LBuff.Clear();
                this._LBuff = new List<byte>(2);
            }

            List<TSocketMessage> list = new List<TSocketMessage>();
            MemoryStream ms = new MemoryStream(buff);
            BinaryReader buffers = new BinaryReader(ms, this.utf8);
            try
            {
                byte[] _buff;
            Label_00983:

                #region 包头读取
                //循环读取包头           
                //判断本次解析的字节是否满足常量字节数 
                if ((buffers.BaseStream.Length - buffers.BaseStream.Position) < ConstLenght)
                {
                    _buff = buffers.ReadBytes((int)(buffers.BaseStream.Length - buffers.BaseStream.Position));
                    this._LBuff.AddRange(_buff);
                    return list;
                }
                Int16 tt1 = buffers.ReadInt16();
                Int16 tt2 = buffers.ReadInt16();
                if (!(tt1 == t1 && tt2 == t2))
                {
                    long ttttt = buffers.BaseStream.Seek(-3, SeekOrigin.Current);
                    goto Label_00983;
                }
                #endregion

                #region 包协议
                int offset = buffers.ReadInt32();
                #endregion

                #region 包解析
                //剩余字节数大于本次需要读取的字节数
                if (offset <= (buffers.BaseStream.Length - buffers.BaseStream.Position))
                {
                    int msgID = buffers.ReadInt32();
                    _buff = buffers.ReadBytes(offset - 4);
                    list.Add(new TSocketMessage(msgID, _buff));
                    if ((buffers.BaseStream.Length - buffers.BaseStream.Position) > 0)
                    {
                        goto Label_00983;
                    }
                }
                else
                {
                    //剩余字节数刚好小于本次读取的字节数 存起来，等待接受剩余字节数一起解析
                    _buff = buffers.ReadBytes((int)(buffers.BaseStream.Length - buffers.BaseStream.Position + ConstLenght));
                    this._LBuff.AddRange(_buff);
                }
                #endregion
            }
            catch (Exception ex) { Console.WriteLine(ex); }
            finally
            {
                if (buffers != null) { buffers.Dispose(); }
                buffers.Close();
                if (buffers != null) { buffers.Dispose(); }
                ms.Close();
                if (ms != null) { ms.Dispose(); }
            }
            return list;
        }
    }
}