unity探索者之socket传输protobuf字节流（三）

              1
            //
            创建消息数据模型

              2
            //
            正式项目中，消息的结构一般是消息长度+消息id+消息主体内容
              3
            public
            class
             Message

              4
            {

              5
            public
             IExtensible protobuf;

              6
            public
            int
             messageId;

              7
            }

              8
              9
            public
            class
             SocketClientTemp : MonoBehaviour

             10
            {

             11
            const
            int packageMaxLength = 1024;
 12 13    Socket mSocket;
 14    Thread threadSend;
 15    Thread threadRecive;
 16     Queue<Message> allMessages = new Queue<Message>();
 17     Queue<byte[]> sendQueue = new Queue<byte[]>();
 18 19publicbool Init()
 20    {
 21//创建一个socket对象 22         mSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
 23return SocketConnection("此处是ip", 1111);
 24    }
 25 26void Update()
 27    {
 28        AnalysisMessage();
 29    }
 30 31///<summary> 32/// 建立服务器连接
 33///</summary> 34///<param name="ip">服务器的ip地址</param> 35///<param name="port">端口</param> 36bool SocketConnection(string ip, int port)
 37    {
 38try 39        {
 40             IPEndPoint ipep = new IPEndPoint(IPAddress.Parse(ip), port);
 41//同步连接服务器，实际使用时推荐使用异步连接，处理方式会在下一篇讲断线重连时讲到 42            mSocket.Connect(ipep);
 43//连接成功后，创建两个线程，分别用于发送和接收消息 44             threadSend = new Thread(new ThreadStart(SendMessage));
 45            threadSend.Start();
 46             threadRecive = new Thread(new ThreadStart(ReceiveMessage));
 47            threadRecive.Start();
 48returntrue;
 49        }
 50catch (Exception e)
 51        {
 52            Debug.Log(e.ToString());
 53            Close();
 54returnfalse;
 55        }
 56    }
 57 58#region ...发送消息
 59///<summary> 60/// 添加数据到发送队列
 61///</summary> 62///<param name="protobufModel"></param> 63///<param name="messageId"></param> 64publicvoid AddSendMessageQueue(IExtensible protobufModel, int messageId)
 65    {
 66        sendQueue.Enqueue(BuildPackage(protobufModel, messageId));
 67    }
 68 69void SendMessage()
 70    {
 71//循环获取发送队列中第一个数据，然后发送到服务器 72while (true)
 73        {
 74if (sendQueue.Count == 0)
 75            {
 76                 Thread.Sleep(100);
 77continue;
 78            }
 79if (!mSocket.Connected)
 80            {
 81                Close();
 82break;
 83            }
 84else 85                 Send(sendQueue.Peek());//发送队列中第一条数据 86        }
 87    }
 88 89void Send(byte[] bytes)
 90    {
 91try 92        {
 93            mSocket.Send(bytes, SocketFlags.None);
 94//发送成功后，从发送队列中移除已发送的消息 95            sendQueue.Dequeue();
 96        }
 97catch (SocketException e)
 98        {
 99//如果错误码为10035，说明服务器缓存区满了，所以等100毫秒再次发送100if (e.NativeErrorCode == 10035)
101            {
102                 Thread.Sleep(100);
103                Send(bytes);
104            }
105else106                Debug.Log(e.ToString());
107        }
108    }
109#endregion110111#region ...接收消息
112///<summary>113/// 解析收到的消息
114///</summary>115void AnalysisMessage()
116    {
117while (allMessages.Count > 0)
118        {
119int id = allMessages.Dequeue().messageId;
120switch (id)
121            {
122//根据消息id做不同的处理123            }
124        }
125    }
126127///<summary>128/// 接收数据
129///</summary>130void ReceiveMessage()
131    {
132while (true)
133        {
134if (!mSocket.Connected)
135break;
136byte[] recvBytesHead = GetBytesReceive(4);
137int bodyLength = IPAddress.NetworkToHostOrder(BitConverter.ToInt32(recvBytesHead, 0));
138byte[] recvBytesBody = GetBytesReceive(bodyLength);
139140byte[] messageId = newbyte[4];
141             Array.Copy(recvBytesBody, 0, messageId, 0, 4);
142byte[] messageBody = newbyte[bodyLength - 4];
143             Array.Copy(recvBytesBody, 4, messageBody, 0, bodyLength - 4);
144145if (BitConverter.IsLittleEndian)
146                Array.Reverse(messageId);
147             FillAllPackages(BitConverter.ToInt32(messageId, 0), messageBody);
148        }
149    }
150151///<summary>152/// 填充接收消息队列
153///</summary>154///<param name="messageId"></param>155///<param name="messageBody"></param>156void FillAllPackages(int messageId, byte[] messageBody)
157    {
158switch (messageId)
159        {
160//根据消息id处理消息，并添加到接收消息队列161case1:
162                 allMessages.Enqueue(new Message() 
163                {
164                     protobuf = ProtobufSerilizer.DeSerialize<TestTemp>(messageBody), 
165                     messageId = messageId 
166                });
167break;
168        }
169    }
170171///<summary>172/// 接收数据并处理
173///</summary>174///<param name="length"></param>175///<returns></returns>176byte[] GetBytesReceive(int length)
177    {
178byte[] recvBytes = newbyte[length];
179while (length > 0)
180        {
181byte[] receiveBytes = newbyte[length < packageMaxLength ? length : packageMaxLength];
182int iBytesBody = 0;
183if (length >= receiveBytes.Length)
184                 iBytesBody = mSocket.Receive(receiveBytes, receiveBytes.Length, 0);
185else186                 iBytesBody = mSocket.Receive(receiveBytes, length, 0);
187             receiveBytes.CopyTo(recvBytes, recvBytes.Length - length);
188             length -= iBytesBody;
189        }
190return recvBytes;
191    }
192#endregion193194///<summary>195/// 构建消息数据包
196///</summary>197///<param name="protobufModel"></param>198///<param name="messageId"></param>199byte[] BuildPackage(IExtensible protobufModel, int messageId)
200    {
201byte[] b;
202if (protobufModel != null)
203             b = ProtobufSerilizer.Serialize(protobufModel);
204else205             b = newbyte[0];
206//消息长度(int数据，长度4) + 消息id(int数据，长度4) + 消息主体内容207         ByteBuffer buf = ByteBuffer.Allocate(b.Length + 4 + 4);
208//消息长度 = 消息主体内容长度 + 消息id长度209         buf.WriteInt(b.Length + 4);
210        buf.WriteInt(messageId);
211212if (protobufModel != null)
213            buf.WriteBytes(b);
214return buf.GetBytes();
215    }
216217void OnDestroy()
218    {
219//停止运行后，如果不关闭socket多线程，再次运行时，unity会卡死220        Close();
221    }
222223///<summary>224/// 关闭socket,终止线程
225///</summary>226publicvoid Close()
227    {
228if (mSocket != null)
229        {
230//微软官方推荐在关闭socket前先shutdown，但是经过测试，发现网络断开后，shutdown会无法执行231if (mSocket.Connected)
232                mSocket.Shutdown(SocketShutdown.Both);
233            mSocket.Close();
234             mSocket = null;
235        }
236//关闭线程237if (threadSend != null)
238            threadSend.Abort();
239if (threadRecive != null)
240            threadRecive.Abort();
241         threadSend = null;
242         threadRecive = null;
243    }
244 }