前言

本章主要介绍和实现怎样正确和高效地处理TCP数据（数据流）。也解决了上一章我们遇到的一些问题

4.1TCP数据流

4.1.1系统缓冲区

收到对端数据时，操作系统会将数据存入到Socket的接收缓冲区中，而且操作系统层面上的缓冲区完全由操作系统操作，程序并不能直接操作，只能通过socket.Receive、socket.Send方法来间接操作。
注意：之后出现的readBuff不是发送缓冲区也不是接收缓冲区，而是用户自定义的缓冲区，用于存放两个操作系统缓冲区读取出的字节数据。

4.1.2粘包半包现象

粘包：
有时候你想发送两条数据“童立华”和“好帅”，期望其他客户端分别展示这两条数据。但是Receive可能把两条信息当做一条处理，最后显示出来的是“童立华好帅”。这就是粘包现象，因为Receive返回操作系统接收缓冲区中存放的内容。
半包：
当你想发送“童立华颜晓仪是猪”，但是接收端调用Receive的时候，只接收到了“童立华”，等待一小段时间后才接收到了“颜晓仪是猪”。最后就分开显示“童立华”，“颜晓仪是猪”两条数据。
因为TCP是基于流的数据，粘包很正常。但是直觉告诉我们：一次发送多少数据，一次就接收多少数据才正常。

在服务端Accept后用

System.Threading.Thread.Sleep(30*1000);

，并在此期间让客户端多次发送数据，就可以人工复现粘包。

4.3解决粘包问题

• 长度信息法：在每个数据包前加上长度信息，每次收到数据后，先读取表示长度的字节，然后从缓冲区取出相应长度的字节。Int16的范围是0-65535，一般一条消息的长度用Int16就可以了。
• 固定长度法：每次读取固定长度的信息，如果有超出的，就取出然后等下次接收信息后拼接。
• 结束符号法：可以用一个结束符号作为消息间的分隔符。当读取到结束符时如果还有消息，就取出等下次接收信息后拼接。

4.3.1发送数据

如果要发送HelloWorld，用长度信息法来解决。最后发送的就是“0AHelloWorld”。用Linq命名空间下的Concat方法来拼接长度数组和信息数组后发送即可。代码最后呈上。

4.3.2接收数据

核心思想是定义一个缓冲区readBuff和记录缓冲区现在有多少数据的长度变量buffCount。如果缓冲区有未处理的数据，就把新读的数据放在有效数据之后。

4.3.3处理数据

如果缓冲区数据足够长，超过一条消息的长度，就把消息提取出来处理。
如果数据长度不够，就不去处理它，等待下一次接受数据。

• 缓冲区长度小于等于2，那就是不够将长度信息解析出来，就等到下一次接受数据。
• 缓冲区长度大于2，但不足以组成一条消息的时候，比如05hell，就不去处理它，等待下一次接收。
• 缓冲区长度大于等于一条完整消息，就解析出来，然后更新缓冲区，也就是用array.copy()函数将后面的移到前面，因为解析完的缓冲区数据已经没用了。

4.4大端小端问题

粘包半包的问题占据了收发数据问题的80%，大端小端问题就是剩余的20%其中之一。
我们是用

BitConverter.ToInt16()

来将长度标记字节转为Int16的，但是看了它的源码会发现，根据计算机是大端还是小端编码，计算编码方式会有不同。
那么对于不同的计算机，读取出来的数据长度也会有不同！

4.4.1为什么有大端小端之分

总而言之是个历史问题，我就不多赘述了！

4.4.2用Reverse()兼容大小端编码

我们规定使用小端编码，就判断系统是否是小端编码的系统，如果不是就用Reverse()将大端编码转为小端。

if(!BitConverter.IsLittleEndian){lenBytes.Reverse();
}

所以接下来我们都是手动还原前两位数字为Int16，用小端编码的还原形式

Int16 bodyLen = (short)((readBuff[1] << 8) | readBuff[0]);

此处的“|”是逻辑与，等同于位相加。

4.5完整发送数据

简单假设我们操作系统缓冲区为8字节

• 先发送04hero，剩下2字节，网络不好，没法送出去，继续在操作系统的发送缓冲区中
• 再发送03cat，cat写不下，03被写入，被成功发送后，缓冲区为空
• 再发送02hi
• 服务器解析时,0302h就被解析出来了，导致出错

4.5.2如何解决发送不完整问题

在发送前将数据保存起来，如果不完整，在Send回调函数中继续发送没发完的数据，直到把新的数据发完才能让新的数据进入发送缓冲区
防止在调用BeginSend调用回调这个时间段内再次点击发送出现问题，我们用一个写入队列是否为空来判断，**每次只能有一条数据！！！**如果还没发干净，你再点击send也不会把新数据加进缓冲区

4.5.3ByteArray和Queue

ByteArray是什么？因为我们要发送没发完的数据，所以就用ByteArray来封装byte[]、readIdx和length，这样才可以用readIdx作为下标记录上一次发送到哪里，直接取出byte[]对应的位数发送即可。
Queue的好处在于入队出队为O(1)，如果是一个数组，那么就需要O(n)的时间复杂度来移动。

4.5.4解决线程冲突

由异步机制可以知道，BeginSend和回调函数执行于不同县城，如果同时操作writeQueue就会出错。比如第二次发送时，第一次发送的回调函数刚好被调用，当第一次出完队之后，再次判断是否要发送，此时第二次发送的东西刚好入队，那么第一次以为是自己没发干净的东西，再发一次。那么第二次又发一次，就将第二次的数据发送了两次。
为了避免线程竞争，可以通过加锁lock的方式处理。当两个线程争夺一个锁的时候，一个线程等待，被组织的那个锁变为可用。

4.6高效的接收数据

4.6.1不足之处

1.Copy操作
O(n)的时间复杂度当然达咩，所以我们用一个ByteArray数组来作为缓冲区，使用readIdx指向缓冲区的第一个数据，每次解析数据后，就将readIdx增加。比如解析了"03cat"，就+5，当缓冲区长度不够时，做一次Array.Copy即可。非常舒服。

2.缓冲区不够长
如果网络不好，就会把缓冲区撑爆。当长度不够时，让它自动扩展，重新申请一个较长的bytes数组。

4.6.2完整的ByteArray

using System.Collections;
using System.Collections.Generic;
using UnityEngine;
using System;public class ByteArray
{//默认大小const int DEFAULT_SIZE = 1024;//初始大小int initSize = 0;//用户缓冲区public byte[] bytes;//读写位置public int readIdx = 0;public int writeIdx = 0;//容量private int capacity = 0;//剩余空间public int remain { get { return capacity - writeIdx; } }//数据长度public int length { get { return writeIdx - readIdx; } }//构造函数public ByteArray(byte[] defaultBytes){bytes = defaultBytes;capacity = defaultBytes.Length;initSize = defaultBytes.Length;readIdx = 0;writeIdx = defaultBytes.Length;}//构造函数public ByteArray(int size = DEFAULT_SIZE){bytes = new byte[size];capacity = size;initSize = size;readIdx = 0;writeIdx = 0;}public override string ToString(){return BitConverter.ToString(bytes, readIdx, length);}public string Debug(){return string.Format("readIdx({0}) writeIdx({1}) bytes({2}))", readIdx, writeIdx, BitConverter.ToString(bytes, 0, bytes.Length));}public void Resize(int size){if (size < length) return;if (size < initSize) return;int n = 1;//n是2的倍数，但是大于size！while (n < size) n *= 2;capacity = n;byte[] newBytes = new byte[capacity];Array.Copy(bytes, readIdx, newBytes, 0, writeIdx - readIdx);bytes = newBytes;writeIdx = length;readIdx = 0;}//检查并移动数据,,多点remain，避免bytes过长public void CheckAndMoveBytes(){if(length < 8){MoveBytes();}}//移动数据public void MoveBytes(){if(length > 0){Array.Copy(bytes, readIdx, bytes, 0, length);}writeIdx = length;readIdx = 0;}//写入数据public int Write(byte[] bs,int offst,int count){if(remain < count){Resize(length + count);}Array.Copy(bs, offst, bytes, writeIdx, count);writeIdx += count;return count;}//读取数据public int Read(byte[] bs, int offset, int count){count = Math.Min(count, length);Array.Copy(bytes, readIdx, bs, offset, count);readIdx += count;CheckAndMoveBytes();return count;}//读取Int16public Int16 ReadInt16(){if (length < 2) return 0;Int16 ret = (Int16)((bytes[readIdx + 1] << 8) | bytes[readIdx]);//手动还原readIdx += 2;CheckAndMoveBytes();return ret;}//读取32public Int32 ReadInt32(){if (length < 4) return 0;Int32 ret = (Int32)((bytes[readIdx + 3] << 24) |(bytes[readIdx + 2] << 16) |(bytes[readIdx + 1] << 8) |(bytes[readIdx + 0]));readIdx += 4;//4位读取完毕，移动下标CheckAndMoveBytes();return ret;}
}

4.7客户端代码

using System.Collections;
using System.Collections.Generic;
using UnityEngine;
using System.Net.Sockets;
using UnityEngine.UI;
using System;
using System.Linq;public class Echo : MonoBehaviour
{//定义套接字Socket socket;//UGUIpublic InputField InputField;public Text text;//接收缓冲区byte[] readBuff = new byte[1024];//最新的接收缓冲区ByteArray readBuff1 = new ByteArray();//接收缓冲区的数据长度int buffCount = 0;//显示文字string recvStr = "";bool canSend = false;List<Socket> checkRead = new List<Socket>();//定义发送缓冲区byte[] sendBytes = new byte[1024];//缓冲区偏移值int readIdx = 0;//缓冲区剩余长度int length = 0;Queue<ByteArray> writeQueue = new Queue<ByteArray>();//点击连接按钮public void Connection(){//新建sockect//地址族IPV4,套接字类型stream,协议类型socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);//connect//远程IP地址,远程端口，阻塞方法，卡住直到服务端回应、//自己建立服务器的话，ip地址和端口号就是这两个socket.Connect("127.0.0.1", 8888);//如果是异步就在callback里接收，同步就直接接收socket.BeginReceive(readBuff1.bytes, readBuff1.writeIdx, readBuff1.remain, 0, ReceiveCallback, socket);//异步//socket.BeginConnect("127.0.0.1", 8888, ConnectCallback, socket);}//Connect回调public void ConnectCallback(IAsyncResult ar){try{//此socket可由ar.AsyncState获取到Socket socket = (Socket)ar.AsyncState;socket.EndConnect(ar);Debug.Log("Socket Connect Succ");//接收缓冲区、0表示从readBuff第0位开始接收数据(和TCP粘包问题有关)、每次最多接收1024字节数据(即使服务器发送1025,也只接收1024)//接收函数调用时机：在连接成功后就开始接受数据，接收到数据后，回调函数ReceiveCallback被调用socket.BeginReceive(readBuff, 0, 1024, 0, ReceiveCallback, socket);}catch (SocketException ex){Debug.Log("Socket Connect fail" + ex.ToString());}}//Receive回调public void ReceiveCallback(IAsyncResult ar){try{Socket socket = (Socket)ar.AsyncState;//获取接收到的数据长度，并更新缓冲区的数据长度int count = socket.EndReceive(ar);readBuff1.writeIdx += count;//处理二进制消息OnReceiveData();//用了专门的处理函数就不需要了//string s = System.Text.Encoding.Default.GetString(readBuff, 0, count);//recvStr = s + "\n" + recvStr;//等待，模拟粘包//System.Threading.Thread.Sleep(1000 * 30);//接收完一串数据后，等待下一串数据的到来if(readBuff1.remain < 8){readBuff1.MoveBytes();readBuff1.Resize(readBuff1.length * 2);}socket.BeginReceive(readBuff1.bytes, readBuff1.writeIdx, readBuff1.remain, 0, ReceiveCallback, socket);}catch (SocketException ex){Debug.Log("Socket Receive fail" + ex.ToString());}}public void OnReceiveData(){Debug.Log("[Recv 1] buffCount=" + readBuff1.length);Debug.Log("[Recv 2] readbuff=" + readBuff1.ToString());//消息长度if (readBuff1.length <= 2) return;int readIdx = readBuff1.readIdx;byte[] bytes = readBuff1.bytes;Int16 bodyLength = (Int16)((bytes[readIdx + 1] << 8)|bytes[readIdx]);//手动以小端形式来还原if (readBuff1.length < 2 + bodyLength) return;readBuff1.readIdx += 2;Debug.Log("[Recv 3] bodyLength=" + bodyLength);//消息体byte[] stringByte = new byte[bodyLength];readBuff1.Read(stringByte, 0, bodyLength);string s = System.Text.Encoding.UTF8.GetString(stringByte);Debug.Log("[Recv 4] s=" + s);//更新缓冲区//int start = 2 + bodyLength;//int count = buffCount - start;//Array.Copy(readBuff, start, readBuff, 0, count);//buffCount -= start;Debug.Log("[Recv 5] readbuff=" + readBuff1.ToString());//更新后的buffcount//消息处理recvStr = s + '\n' + recvStr;//继续读取if (readBuff1.length > 2){OnReceiveData();}}//点击发送按钮public void Send(){//send//if (canSend)//{//    string sendStr = InputField.text;//    //string sendStr = System.DateTime.Now.ToString();//    //将str转化为字节流//    byte[] sendBytes = System.Text.Encoding.Default.GetBytes(sendStr);//    //socket.BeginSend(sendBytes, 0, sendBytes.Length, 0, SendCallback, socket);//    socket.Send(sendBytes);//}//异步不需要receive//Recv//byte[] readBuff = new byte[1024];接收数据的长度//int count = socket.Receive(readBuff);//string recvStr = System.Text.Encoding.Default.GetString(readBuff, 0, count);//text.text = recvStr;Close//socket.Close();string sendStr = InputField.text;//组装协议byte[] bodyBytes = System.Text.Encoding.Default.GetBytes(sendStr);Int16 len = (Int16)bodyBytes.Length;byte[] lenBytes = BitConverter.GetBytes(len);//大小端编码if (!BitConverter.IsLittleEndian){Debug.Log("[Send] Reverse lenBytes");lenBytes.Reverse();}byte[] sendBytes = lenBytes.Concat(bodyBytes).ToArray();//现在开始用ByteArray来封装这些数据//length = sendBytes.Length;//数据长度//readIdx = 0;方便：同步、不抛异常//socket.BeginSend(sendBytes, readIdx, length, 0, SendCallback, socket);//Debug.Log("[Send]" + BitConverter.ToString(sendBytes));ByteArray ba = new ByteArray(sendBytes);int count = 0;//加锁，只有一个线程可以操作lock (writeQueue){writeQueue.Enqueue(ba);count = writeQueue.Count;}//一定要把前面的发完，只剩下当前要发送的才发送if(count == 1){socket.BeginSend(ba.bytes, ba.readIdx, ba.length, 0, SendCallback, socket);}}//Send回调public void SendCallback(IAsyncResult ar){try{//这个socket是传进回调的用户定义对象，可强转为socketSocket socket = (Socket)ar.AsyncState;int count = socket.EndSend(ar);//判断是否发送完整ByteArray ba;//加锁！lock (writeQueue){ba = writeQueue.First();}//每次加上发送的长度，看看end - start = length是否为0，为0证明发完就可以弹出ba.readIdx += count;if (ba.length == 0){//只要是取first的都要加锁！lock (writeQueue){//如果剩余长度为0，证明发送完整了writeQueue.Dequeue();ba = writeQueue.First();}}//但是如果发送不完整，是不是会收到一条完整的和一条不完整的？//不是，会先发出一段，再发出后半段。因为readIdx也就是start在变if(ba != null){//如果发送不完整或者发送完整且存在第二条数据socket.BeginSend(ba.bytes, ba.readIdx, ba.length, 0, SendCallback, socket);}Debug.Log("Socket Send succ" + count);}catch(SocketException ex){Debug.Log("Socket Send fail" + ex.ToString());}}public void Update(){//if(socket == null)//{//    return;//}poll客户端//if (socket.Poll(0, SelectMode.SelectRead))//{//    byte[] readBuff = new byte[1024];//    int count = socket.Receive(readBuff);//    //不阻塞模式，microSeconds=0//    string recvStr = System.Text.Encoding.Default.GetString(readBuff, 0, count);//    text.text = recvStr;//}处理阻塞send应该也差不多//if (socket.Poll(0, SelectMode.SelectWrite))//{//    canSend = true;//}//else//{//    canSend = false;//}select客户端只需检测一个socket，将这个socket加入到待监测列表即可//checkRead.Clear();//checkRead.Add(socket);select//Socket.Select(checkRead, null, null, 0);check//foreach (Socket s in checkRead)//{//    byte[] readBuff = new byte[1024];//    int count = socket.Receive(readBuff);//    string recvStr = System.Text.Encoding.Default.GetString(readBuff, 0, count);//    text.text = recvStr;//}text.text = recvStr;}
}

4.8服务端代码

using System;
using System.Net;
using System.Net.Sockets;
using System.Collections.Generic;
using System.Reflection;
using System.Linq;//hhpublic class ClientState
{public Socket socket;public byte[] readBuff = new byte[1024];public int hp = -100;public float x = 0;public float y = 0;public float z = 0;public float eulY = 0;
}
class MainClass
{//异步服务器//监听Socketstatic Socket listenfd;//客户端Socket及状态信息public static Dictionary<Socket, ClientState> clients = new Dictionary<Socket, ClientState>();public static void Main(string[] args){Console.WriteLine("Hello");//SocketSocket listenfd = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);//BindIPAddress ipAdr = IPAddress.Parse("127.0.0.1");//IP地址IPEndPoint ipEp = new IPEndPoint(ipAdr, 8888);//IP和端口//给listenfd套接字绑定IP和端口,"127.0.0.1"地址和8888号端口listenfd.Bind(ipEp);//Listen，等待客户端连接//0表示容纳等待接收的客户端连接数不受限制，1代表最多可容纳等待接受的连接数为1listenfd.Listen(0);Console.WriteLine("[服务器]启动成功");//同步服务器//while (true)//{//    //Accept//    //接收客户端连接，均为阻塞方法，如果没有客户端连接就不会向下执行//    //返回一个新客户端的socket对象，对于服务器来说//    //它有一个监听 Socket(listenfd)用来监听和应答客户端的连接，对每个客户端还有专门一个socket(connfd)用于处理客户端的数据//    Socket connfd = listenfd.Accept();//    Console.WriteLine("[服务器]Accetp");//    //Receive//    byte[] readBuff = new byte[1024];//    int count = connfd.Receive(readBuff);//    string readStr = System.Text.Encoding.Default.GetString(readBuff, 0, count);//    Console.WriteLine("[服务器接收]" + readStr);//    //Send//    byte[] sendBytes = System.Text.Encoding.Default.GetBytes(readStr);//    //如果客户端用异步，即使收不到服务端的消息也不会卡住主线程//    connfd.Send(sendBytes);//}//异步服务器//listenfd.BeginAccept(AcceptCallback, listenfd);等待//Console.ReadLine();阻塞poll服务器//while (true)//{//    //检查listenfd，可读就添加客户端信息//    if (listenfd.Poll(0, SelectMode.SelectRead))//    {//        ReadListenfd(listenfd);//    }//    //检查clientfd//    foreach (ClientState s in clients.Values)//    {//        Socket clientfd = s.socket;//        if (clientfd.Poll(0, SelectMode.SelectRead))//        {//            //false表示客户端断开(收到长度为0的数据)//            //断开会删掉列表中对应的信息，导致遍历失败，所以直接break//            if (!ReadClientfd(clientfd))//            {//                break;//            }//        }//    }//    //防止CPU占用过高//    //让程序挂起1ms，避免死循环让CPU喘息//    System.Threading.Thread.Sleep(1);//}//select服务器//checkReadList<Socket> checkRead = new List<Socket>();//主循环while (true){//填充checkRead列表checkRead.Clear();checkRead.Add(listenfd);foreach (ClientState s in clients.Values){checkRead.Add(s.socket);}//select，只将待检查可读的列表传进去Socket.Select(checkRead, null, null, 1000);//调用完上面的方法后这个列表就被改了，这个列表中只有可读的socketforeach (Socket s in checkRead){//因为listnfd本身就被加进去了if (s == listenfd){ReadListenfd(s);}//除了listen其余都是可读的客户端，直接处理即可else{ReadClientfd(s);}}}
}//读取listenfd，和一步服务端的acceptcallback相似，用于应答客户端，添加客户端信息public static void ReadListenfd(Socket listenfd){Console.WriteLine("Accept");Socket clientfd = listenfd.Accept();ClientState state = new ClientState();state.socket = clientfd;clients.Add(clientfd, state);}//和异步服务端的Receivecallback类似，用于接收客户端消息，并广播给所有客户端public static bool ReadClientfd(Socket clientfd){ClientState state = clients[clientfd];//接收int count = 0;try{count = clientfd.Receive(state.readBuff);}catch(SocketException ex){clientfd.Close();clients.Remove(clientfd);Console.WriteLine("Receive SocketException" + ex.ToString());return false;}//让客户端关闭if(count == 0){clientfd.Close();clients.Remove(clientfd);Console.WriteLine("Socket Close");return false;}//广播//enter|和list|一起到了怎么拆分，因为是同一个客户端发过来的，都在readbuff里string recvStr = System.Text.Encoding.Default.GetString(state.readBuff, 2, count-2);Console.WriteLine("Receive" + recvStr);byte[] sendBytes = new byte[count];Array.Copy(state.readBuff, 0, sendBytes, 0, count);foreach (ClientState cs in clients.Values){cs.socket.Send(sendBytes);}return true;}//Accept回调//是beginaccept的回调函数，处理3件事//1.给新的连接分配ClientState，并把它加入到clients列表中//2.异步接收客户端数据//3.再次调用BeginAccept循环public static void AcceptCallback(IAsyncResult ar){try{Console.WriteLine("[服务器]Accept");//监听和应答客户端的socketSocket listenfd = (Socket)ar.AsyncState;//处理该客户端的socketSocket clientfd = listenfd.EndAccept(ar);//clients列表ClientState state = new ClientState();//初始化此客户端类，key和value岂不是重复利用了？state.socket = clientfd;clients.Add(clientfd, state);//接收数据BeginReceive，以ClientState取代Socketclientfd.BeginReceive(state.readBuff, 0, 1024, 0, ReceiveCallback, state);//继续Acceptlistenfd.BeginAccept(AcceptCallback, listenfd);}catch(SocketException ex){Console.WriteLine("Socket Accept fail" + ex.ToString());}}//Receive回调//1.服务端收到消息后，回应客户端//2.如果收到客户端关闭连接的信号"if(count==0)"，断开连接//3.继续调用BeginReceive接收下一个数据public static void ReceiveCallback(IAsyncResult ar){try{//发送消息的客户端ClientState state = (ClientState)ar.AsyncState;Socket clientfd = state.socket;//当receive返回值小于等于0时，表示socket连接可以断开int count = clientfd.EndReceive(ar);//客户端关闭if(count == 0){clientfd.Close();clients.Remove(clientfd);Console.WriteLine("Socket Close");return;}//从收到的字节流转为stringstring recvStr = System.Text.Encoding.Default.GetString(state.readBuff, 0, count);string sendStr = clientfd.RemoteEndPoint.ToString() + ":" + recvStr;byte[] sendBytes = System.Text.Encoding.Default.GetBytes(sendStr);//string转为bytes//用于处理该客户端数据的socketforeach(ClientState s in clients.Values){s.socket.Send(sendBytes);}clientfd.BeginReceive(state.readBuff, 0, 1024, 0, ReceiveCallback, state);}catch(SocketException ex){Console.WriteLine("Socket Receive fail" + ex.ToString());}}public static void Send(ClientState cs,string sendStr){byte[] sendBytes = System.Text.Encoding.Default.GetBytes(sendStr);cs.socket.Send(sendBytes);}
}