对于基于TCP开发的通讯程序,有个很重要的问题需要解决,就是封包和拆包.自从我从事网络通讯编程工作以来(大概有三年的时间了),我一直在思索和改进封包和拆包的方法.下面就针对这个问题谈谈我的想法,抛砖引玉.若有不对,不妥之处,恳求大家指正.在此先谢过大家了.
一.为什么基于TCP的通讯程序需要进行封包和拆包.
TCP是个"流"协议,所谓流,就是没有界限的一串数据.大家可以想想河里的流水,是连成一片的,其间是没有分界线的.但一般通讯程序开发是需要定义一个 个相互独立的数据包的,比如用于登陆的数据包,用于注销的数据包.由于TCP"流"的特性以及网络状况,在进行数据传输时会出现以下几种情况.
假设我们连续调用两次send分别发送两段数据data1和data2,在接收端有以下几种接收情况(当然不止这几种情况,这里只列出了有代表性的情况).
A.先接收到data1,然后接收到data2.
B.先接收到data1的部分数据,然后接收到data1余下的部分以及data2的全部.
C.先接收到了data1的全部数据和data1的部分数据,然后接收到了data2的余下的数据.
D.一次性接收到了data1和data2的全部数据.
对于A这种情况正是我们需要的,不再做讨论.对于B,C,D的情况就是大家经常说的"粘包",就需要我们把接收到的数据进行拆包,拆成一个个独立的数据包.为了拆包就必须在发送端进行封包.
另:对于UDP来说就不存在拆包的问题,因为UDP是个"数据包"协议,也就是两段数据间是有界限的,在接收端要么接收不到数据要么就是接收一个完整的一段数据,不会少接收也不会多接收.
二.为什么会出现B.C.D的情况.
"粘包"可发生在发送端也可发生在接收端.
1.由Nagle算法造成的发送端的粘包:Nagle算法是一种改善网络传输效率的算法.简单的说,当我们提交一段数据给TCP发送时,TCP并不立刻发 送此段数据,而是等待一小段时间,看看在等待期间是否还有要发送的数据,若有则会一次把这两段数据发送出去.这是对Nagle算法一个简单的解释,详细的 请看相关书籍.象C和D的情况就有可能是Nagle算法造成的.
2.接收端接收不及时造成的接收端粘包:TCP会把接收到的数据存在自己的缓冲区中,然后通知应用层取数据.当应用层由于某些原因不能及时的把TCP的数据取出来,就会造成TCP缓冲区中存放了几段数据.
三.怎样封包和拆包.
最初遇到"粘包"的问题时,我是通过在两次send之间调用sleep来休眠一小段时间来解决.这个解决方法的缺点是显而易见的,使传输效率大大降低,而 且也并不可靠.后来就是通过应答的方式来解决,尽管在大多数时候是可行的,但是不能解决象B的那种情况,而且采用应答方式增加了通讯量,加重了网络负荷. 再后来就是对数据包进行封包和拆包的操作.
封包:
封包就是给一段数据加上包头,这样一来数据包就分为包头和包体两部分内容了(以后讲过滤非法包时封包会加入"包尾"内容).包头其实上是个大小固定的结构 体,其中有个结构体成员变量表示包体的长度,这是个很重要的变量,其他的结构体成员可根据需要自己定义.根据包头长度固定以及包头中含有包体长度的变量就 能正确的拆分出一个完整的数据包.
对于拆包目前我最常用的是以下两种方式.
1.动态缓冲区暂存方式.之所以说缓冲区是动态的是因为当需要缓冲的数据长度超出缓冲区的长度时会增大缓冲区长度.
大概过程描述如下:
A,为每一个连接动态分配一个缓冲区,同时把此缓冲区和SOCKET关联,常用的是通过结构体关联.
B,当接收到数据时首先把此段数据存放在缓冲区中.
C,判断缓存区中的数据长度是否够一个包头的长度,如不够,则不进行拆包操作.
D,根据包头数据解析出里面代表包体长度的变量.
E,判断缓存区中除包头外的数据长度是否够一个包体的长度,如不够,则不进行拆包操作.
F,取出整个数据包.这里的"取"的意思是不光从缓冲区中拷贝出数据包,而且要把此数据包从缓存区中删除掉.删除的办法就是把此包后面的数据移动到缓冲区的起始地址.
这种方法有两个缺点.1.为每个连接动态分配一个缓冲区增大了内存的使用.2.有三个地方需要拷贝数据,一个地方是把数据存放在缓冲区,一个地方是把完整的数据包从缓冲区取出来,一个地方是把数据包从缓冲区中删除.第二种拆包的方法会解决和完善这些缺点.
下面给出相关代码.
前面提到过这种方法的缺点.下面给出一个改进办法, 即采用环形缓冲.但是这种改进方法还是不能解决第一个缺点以及第一个数据拷贝,只能解决第三个地方的数据拷贝(这个地方是拷贝数据最多的地方).第2种拆包方式会解决这两个问题.
环形缓冲实现方案是定义两个指针,分别指向有效数据的头和尾.在存放数据和删除数据时只是进行头尾指针的移动.
用代码来说明.注:下面的代码是采用一个开源的游戏服务器的代码,我对此代码有所修改. 2.利用底层的缓冲区来进行拆包
由于TCP也维护了一个缓冲区,所以我们完全可以利用TCP的缓冲区来缓存我们的数据,这样一来就不需要为每一个连接分配一个缓冲区了.另一方面我们知道
recv或者wsarecv都有一个参数,用来表示我们要接收多长长度的数据.利用这两个条件我们就可以对第一种方法进行优化.
对于阻塞SOCKET来说,我们可以利用一个循环来接收包头长度的数据,然后解析出代表包体长度的那个变量,再用一个循环来接收包体长度的数据.
拆包一直是个硬伤呀，MLGB的，服务端各种乱数据，果断整理下
拆包思路：设计一个网络协议，一般都会分包，一个包就相当于一个逻辑上的命令。
1、如果我们用udp协议，省事的多，一次会收到一个完整的包，但UDP不可靠，顺序也不能保证，当然像QQ对UDP封装的很好，模拟了TCP的可靠性。网上也有一些封装好的可靠的UDP组件，大家用的话可以找找。关于用什么协议好这个问题，本贴不讨论。 2、如果我们用TCP协议不是长连接，像HTTP（不考虑KeepAlive）那样，一个连接上只发送一个包，我们也会很清晰的区分出接受到的每一个包。 3、还有就是我们还用TCP长连接，但每次发送固定长度的包，如果要发送的数据长度不够就用补齐，如果大于固定长度，就分成几个发，这个也很简单实用。 4、再有就是一个包有特定的开始和结尾，比如包头是<bof>包尾是<eof>，我们在可以从头读到尾，并把一个一个的包放入队列，由处理线程去处理。 5、再有一种就是每个包有固定长度的header，这个header里包含一个包的长度信息，我们可以先从头里读出长度信息，然后再借着读这么长的数据，完了这就是一个包。 关于封包的几种类型我就想到这么多，其中的利弊大家一看便知，我就不忽悠了，本文主要介绍最后一种方式，好多网络协议用的都是这种，包括CMPP协议，我们自己设计协议的时候一般不用像CMPP协议那样，因为二进制协议虽然虽然节省网络流量，但可读性不好。出问题，抓个包分析起来太麻烦。我们可以用.net自带的序列功能把要发送的类序列化成XML字符串发送出去，这多好看呀。 由于Socket缓冲区设置及其他的原因，Socket在接受数据的时候有时候不能完整的收到一个包，就是你读出包的长度后，可能不能一次就读取这么多数据。而如果读个半截儿的包就用UTF8Encoding等来解析，会解析出乱码的，我们这里用Encoding.UTF8.GetDecoder()来对包进行成块儿的解析，它就是用来做这种事情的。 下面就来看一下代码，代码的注释很全，演示了一个包从发到接受、解析的全过程，其中接受的过程没有一次收全所有的包，而是收了好几次，但我们最终还是成功的解析了收到的包。public static void UnPack() { //1、声明通过socket发送的字符串 string toSendStringBySocket = "娃娃士大夫%#￥%My name is 蛙蛙王子！！"; //2、转换成utf-8字节数组 byte[] bsInput = Encoding.UTF8.GetBytes(toSendStringBySocket); //3、计算要发送的字节数组的长度，并写到第一块儿字节数组的开头 //一般协议设计里都有一个长度的Header,这里就是写这个Header int inputBytesCount = bsInput.Length; byte[] bs1 = new byte[4 + 3]; //4是一个int的长度，3是底一块字节数组除了Header剩余的大小 Buffer.BlockCopy(BitConverter.GetBytes(inputBytesCount), 0, bs1, 0, 4); //4、把要发送的字节数组拆分成3块儿发出去，因为socket在接受字节数组的时候 //也可能半截半截儿的接收，我们就是要模拟这种效果下的拆包，因为第一块包写了 //一个4个字节的Header，而第一块字节数组长度是7，所以再写三个字节长度的数据 int offSet = 0; Buffer.BlockCopy(bsInput, offSet, bs1, 4, 3); offSet += bs1.Length - 4; //5、写第二块儿数据 byte[] bs2 = new byte[8]; Buffer.BlockCopy(bsInput, offSet, bs2, 0, bs2.Length); offSet += bs2.Length; //6、写第三块儿数据，我们这里模拟在最后一块数据的末尾加一些乱七八糟的数据 //这些乱七八糟的数据有可能是下一个包的header。 byte[] bs3 = new byte[bsInput.Length - offSet + 4]; Buffer.BlockCopy(bsInput, offSet, bs3, 0, bsInput.Length - offSet); Buffer.BlockCopy(new byte[] { 1, 2, 3, 4 }, 0, bs3, bs3.Length - 4, 4); //7、Socket的接收方在执行BeginReceive函数，并回调函数里把收到的数据放入一个队列里 //dotNet的队列内部就是一个环形数组，这里直接就当环形缓冲区来用了。 Queue<byte[]> bufferPool = new Queue<byte[]>(); bufferPool.Enqueue(bs1); bufferPool.Enqueue(bs2); bufferPool.Enqueue(bs3); //8、初始化一些变量准备解包 //声明一个字符串缓冲区，大小是你的协议里规定的最大的包体长度 char[] chars = new char[256]; //定义一个UTF-8的Decoder，它可以成块的解包，内部自动维护解析状态 //关于它的使用请参考MSDN或者《.net框架设计》 Decoder d = Encoding.UTF8.GetDecoder(); int charLen = 0; //定义每次解包返回的字符长度 int parseBytesCount = 0; //定义已解包的字节数 int LenghHeader = 0; //定义收到包的长度 bool needReadLengthHeader = true; //是否需要读取长度的头 int srcOffSet = 0; //定义要解析的数据块的偏移量 byte[] tempBuffer; //9、当环形缓冲里有数据的时候就一直解析 while (bufferPool.Count > 0) { //10、读取数据包的长度信息，LengthHeader //因为第一块儿包包含长度信息，所以要先读出来 //读了长度包后，要把数据库解析偏移量加4 if(needReadLengthHeader) { LenghHeader = BitConverter.ToInt32(bs1, parseBytesCount); needReadLengthHeader = false; srcOffSet = 4; } //11、从环形缓冲区取出一块儿数据 tempBuffer = bufferPool.Dequeue(); parseBytesCount += tempBuffer.Length-srcOffSet; //更改已解析的字节数 //12、如果已解析的字节数大于数据的长度，那么只解需要解析的字节 if (parseBytesCount > LenghHeader) { parseBytesCount -= tempBuffer.Length; d.GetChars(tempBuffer, srcOffSet, inputBytesCount - parseBytesCount, chars, charLen); //这里记录下当前的临时缓冲区已解析到了什么位置，准备解析下一个包 srcOffSet = inputBytesCount - parseBytesCount; // break; } //13、解析这半拉包 charLen += d.GetChars(tempBuffer, srcOffSet, tempBuffer.Length-srcOffSet, chars, charLen); srcOffSet = 0; } string s = new string(chars); //14、通知包处理线程来处理这个包 Console.WriteLine(s); }
实际场景中应用
using UnityEngine; using System.Collections; using System; using System.Threading; using System.Text; using System.Net; using System.Net.Sockets; using System.Collections.Generic; using System.IO; using System.Runtime.InteropServices; using System.Runtime.Serialization; using System.Runtime.Serialization.Formatters.Binary; public class JFSocket { //Socket客户端对象 private Socket clientSocket; //JFPackage.WorldPackage是我封装的结构体， //在与服务器交互的时候会传递这个结构体 //当客户端接到到服务器返回的数据包时，我把结构体add存在链表中。 public List<JFPackage.WorldPackage> worldpackage; //单例模式 private static JFSocket instance; public static JFSocket GetInstance() { if (instance == null) { instance = new JFSocket(); } return instance; } //单例的构造函数 JFSocket() { //创建Socket对象， 这里我的连接类型是TCP clientSocket = new Socket (AddressFamily.InterNetwork,SocketType.Stream,ProtocolType.Tcp); //服务器IP地址 IPAddress ipAddress = IPAddress.Parse ("192.168.1.100"); //服务器端口 IPEndPoint ipEndpoint = new IPEndPoint (ipAddress, 10060); //这是一个异步的建立连接，当连接建立成功时调用connectCallback方法 IAsyncResult result = clientSocket.BeginConnect (ipEndpoint,new AsyncCallback (connectCallback),clientSocket); //这里做一个超时的监测，当连接超过5秒还没成功表示超时 bool success = result.AsyncWaitHandle.WaitOne( 5000, true ); if ( !success ) { //超时 Closed(); Debug.Log("connect Time Out"); }else { //与socket建立连接成功，开启线程接受服务端数据。 worldpackage = new List<JFPackage.WorldPackage>(); Thread thread = new Thread(new ThreadStart(ReceiveSorket)); thread.IsBackground = true; thread.Start(); } } private void connectCallback(IAsyncResult asyncConnect) { Debug.Log("connectSuccess"); } private void ReceiveSorket() { //在这个线程中接受服务器返回的数据 while (true) { if(!clientSocket.Connected) { //与服务器断开连接跳出循环 Debug.Log("Failed to clientSocket server."); clientSocket.Close(); break; } try { //接受数据保存至bytes当中 byte[] bytes = new byte[4096]; //Receive方法中会一直等待服务端回发消息 //如果没有回发会一直在这里等着。 int i = clientSocket.Receive(bytes); if(i <= 0) { clientSocket.Close(); break; } //这里条件可根据你的情况来判断。 //因为我目前的项目先要监测包头长度， //我的包头长度是2，所以我这里有一个判断 if(bytes.Length > 2) { SplitPackage(bytes,0); }else { Debug.Log("length is not > 2"); } } catch (Exception e) { Debug.Log("Failed to clientSocket error." + e); clientSocket.Close(); break; } } } private void SplitPackage(byte[] bytes , int index) { //在这里进行拆包，因为一次返回的数据包的数量是不定的 //所以需要给数据包进行查分。 while(true) { //包头是2个字节 byte[] head = new byte[2]; int headLengthIndex = index + 2; //把数据包的前两个字节拷贝出来 Array.Copy(bytes,index,head,0,2); //计算包头的长度 short length = BitConverter.ToInt16(head,0); //当包头的长度大于0 那么需要依次把相同长度的byte数组拷贝出来 if(length > 0) { byte[] data = new byte[length]; //拷贝出这个包的全部字节数 Array.Copy(bytes,headLengthIndex,data,0,length); //把数据包中的字节数组强制转换成数据包的结构体 //BytesToStruct()方法就是用来转换的 //这里需要和你们的服务端程序商量， JFPackage.WorldPackage wp = new JFPackage.WorldPackage(); wp = (JFPackage.WorldPackage)BytesToStruct(data,wp.GetType()); //把每个包的结构体对象添加至链表中。 worldpackage.Add(wp); //将索引指向下一个包的包头 index = headLengthIndex + length; }else { //如果包头为0表示没有包了，那么跳出循环 break; } } } //向服务端发送一条字符串 //一般不会发送字符串 应该是发送数据包 public void SendMessage(string str) { byte[] msg = Encoding.UTF8.GetBytes(str); if(!clientSocket.Connected) { clientSocket.Close(); return; } try { //int i = clientSocket.Send(msg); IAsyncResult asyncSend = clientSocket.BeginSend (msg,0,msg.Length,SocketFlags.None,new AsyncCallback (sendCallback),clientSocket); bool success = asyncSend.AsyncWaitHandle.WaitOne( 5000, true ); if ( !success ) { clientSocket.Close(); Debug.Log("Failed to SendMessage server."); } } catch { Debug.Log("send message error" ); } } //向服务端发送数据包，也就是一个结构体对象 public void SendMessage(object obj) { if(!clientSocket.Connected) { clientSocket.Close(); return; } try { //先得到数据包的长度 short size = (short)Marshal.SizeOf(obj); //把数据包的长度写入byte数组中 byte [] head = BitConverter.GetBytes(size); //把结构体对象转换成数据包，也就是字节数组 byte[] data = StructToBytes(obj); //此时就有了两个字节数组，一个是标记数据包的长度字节数组， 一个是数据包字节数组， //同时把这两个字节数组合并成一个字节数组 byte[] newByte = new byte[head.Length + data.Length]; Array.Copy(head,0,newByte,0,head.Length); Array.Copy(data,0,newByte,head.Length, data.Length); //计算出新的字节数组的长度 int length = Marshal.SizeOf(size) + Marshal.SizeOf(obj); //向服务端异步发送这个字节数组 IAsyncResult asyncSend = clientSocket.BeginSend (newByte,0,length,SocketFlags.None,new AsyncCallback (sendCallback),clientSocket); //监测超时 bool success = asyncSend.AsyncWaitHandle.WaitOne( 5000, true ); if ( !success ) { clientSocket.Close(); Debug.Log("Time Out !"); } } catch (Exception e) { Debug.Log("send message error: " + e ); } } //结构体转字节数组 public byte[] StructToBytes(object structObj) { int size = Marshal.SizeOf(structObj); IntPtr buffer = Marshal.AllocHGlobal(size); try { Marshal.StructureToPtr(structObj,buffer,false); byte[] bytes = new byte[size]; Marshal.Copy(buffer, bytes,0,size); return bytes; } finally { Marshal.FreeHGlobal(buffer); } } //字节数组转结构体 public object BytesToStruct(byte[] bytes, Type strcutType) { int size = Marshal.SizeOf(strcutType); IntPtr buffer = Marshal.AllocHGlobal(size); try { Marshal.Copy(bytes,0,buffer,size); return Marshal.PtrToStructure(buffer, strcutType); } finally { Marshal.FreeHGlobal(buffer); } } private void sendCallback (IAsyncResult asyncSend) { } //关闭Socket public void Closed() { if(clientSocket != null && clientSocket.Connected) { clientSocket.Shutdown(SocketShutdown.Both); clientSocket.Close(); } clientSocket = null; } }
为了与服务端达成默契，判断数据包是否完成。我们需要在数据包中定义包头 ，包头一般是这个数据包的长度，也就是结构体对象的长度。正如代码中我们把两个数据类型 short 和 object 合并成一个新的字节数组。
然后是数据包结构体的定义，需要注意如果你在做IOS和Android的话数据包中不要包含数组，不然在结构体转换byte数组的时候会出错。
Marshal.StructureToPtr () error : Attempting to JIT compile method
JFPackage.cs
using UnityEngine; using System.Collections; using System.Runtime.InteropServices; public class JFPackage { //结构体序列化 [System.Serializable] //4字节对齐 iphone 和 android上可以1字节对齐 [StructLayout(LayoutKind.Sequential, Pack = 4)] public struct WorldPackage { public byte mEquipID; public byte mAnimationID; public byte mHP; public short mPosx; public short mPosy; public short mPosz; public short mRosx; public short mRosy; public short mRosz; public WorldPackage(short posx,short posy,short posz, short rosx, short rosy, short rosz,byte equipID,byte animationID,byte hp) { mPosx = posx; mPosy = posy; mPosz = posz; mRosx = rosx; mRosy = rosy; mRosz = rosz; mEquipID = equipID; mAnimationID = animationID; mHP = hp; } }; }
让角色发生移动的时候，调用该方法向服务端发送数据。
oid SendPlayerWorldMessage() { //组成新的结构体对象，包括主角坐标旋转等。 Vector3 PlayerTransform = transform.localPosition; Vector3 PlayerRotation = transform.localRotation.eulerAngles; //用short的话是2字节，为了节省包的长度。这里乘以100 避免使用float 4字节。当服务器接受到的时候小数点向前移动两位就是真实的float数据 short px = (short)(PlayerTransform.x*100); short py = (short)(PlayerTransform.y*100); short pz = (short)(PlayerTransform.z*100); short rx = (short)(PlayerRotation.x*100); short ry = (short)(PlayerRotation.y*100); short rz = (short)(PlayerRotation.z*100); byte equipID = 1; byte animationID =9; byte hp = 2; JFPackage.WorldPackage wordPackage = new JFPackage.WorldPackage(px,py,pz,rx,ry,rz,equipID,animationID,hp); //通过Socket发送结构体对象 mJFsorket.SendMessage(wordPackage); }
接着就是客户端同步服务器的数据，目前是测试阶段所以写的比较简陋，不过原理都是一样的
//上次同步时间 private float mSynchronous; void Update () { mSynchronous +=Time.deltaTime; //在Update中每0.5s的时候同步一次 if(mSynchronous > 0.5f) { int count = mJFsorket.worldpackage.Count; //当接受到的数据包长度大于0 开始同步 if(count > 0) { //遍历数据包中 每个点的坐标 foreach(JFPackage.WorldPackage wp in mJFsorket.worldpackage) { float x = (float)(wp.mPosx / 100.0f); float y = (float)(wp.mPosy /100.0f); float z = (float)(wp.mPosz /100.0f); Debug.Log("x = " + x + " y = " + y+" z = " + z); //同步主角的新坐标 mPlayer.transform.position = new Vector3 (x,y,z); } //清空数据包链表 mJFsorket.worldpackage.Clear(); } mSynchronous = 0; } }