粘包
使用TCP长连接就会引入粘包的问题,粘包是指发送方发送的若干包数据到接收方接收时粘成一包,从接收缓冲区看,后一包数据的头紧接着前一包数据的尾。粘包可能由发送方造成,也可能由接收方造成。TCP为提高传输效率,发送方往往要收集到足够多的数据后才发送一包数据,造成多个数据包的粘连。如果接收进程不及时接收数据,已收到的数据就放在系统接收缓冲区,用户进程读取数据时就可能同时读到多个数据包。
粘包一般的解决办法是制定通讯协议,由协议来规定如何分包解包。
分包
在NETIOCPDemo例子程序中,我们分包的逻辑是先发一个长度,然后紧接着是数据包内容,这样就可以把每个包分开。
应用层数据包格式如下:
AsyncSocketInvokeElement分包处理主要代码,我们收到的数据都是在ProcessReceive方法中处理,处理的方法是把收到的数据存到缓冲区数组中,然后取前4个字节为长度,如果剩下的字节数大于等于长度,则取到一个完整包,进行后续逻辑处理,如果取到的不够一个包,则不处理,等待后续包接收,具体代码如下:
应用层数据包格式 数据包长度Len:Cardinal(4字节无符号整数) 数据包内容,长度为Len
解包public virtual bool ProcessReceive(byte[] buffer, int offset, int count) //接收异步事件返回的数据,用于对数据进行缓存和分包 { m_activeDT = DateTime.UtcNow; DynamicBufferManager receiveBuffer = m_asyncSocketUserToken.ReceiveBuffer; receiveBuffer.WriteBuffer(buffer, offset, count); if (receiveBuffer.DataCount > sizeof(int)) { //按照长度分包 int packetLength = BitConverter.ToInt32(receiveBuffer.Buffer, 0); //获取包长度 if (NetByteOrder) packetLength = System.Net.IPAddress.NetworkToHostOrder(packetLength); //把网络字节顺序转为本地字节顺序 if ((packetLength > 10 * 1024 * 1024) | (receiveBuffer.DataCount > 10 * 1024 * 1024)) //最大Buffer异常保护 return false; if ((receiveBuffer.DataCount - sizeof(int)) >= packetLength) //收到的数据达到包长度 { bool result = ProcessPacket(receiveBuffer.Buffer, sizeof(int), packetLength); if (result) receiveBuffer.Clear(packetLength + sizeof(int)); //从缓存中清理 return result; } else { return true; } } else { return true; } }
由于我们应用层数据包既可以传命令也可以传数据,因而针对每个包我们进行解包,分出命令和数据分别处理,因而每个Socket服务对象都需要解包,我们解包的逻辑是放在ProcessPacket中,命令和数据的包格式为:
命令长度Len:Cardinal(4字节无符号整数) 命令 数据 public virtual bool ProcessPacket(byte[] buffer, int offset, int count) //处理分完包后的数据,把命令和数据分开,并对命令进行解析 { if (count < sizeof(int)) return false; int commandLen = BitConverter.ToInt32(buffer, offset); //取出命令长度 string tmpStr = Encoding.UTF8.GetString(buffer, offset + sizeof(int), commandLen); if (!m_incomingDataParser.DecodeProtocolText(tmpStr)) //解析命令 return false; return ProcessCommand(buffer, offset + sizeof(int) + commandLen, count - sizeof(int) - commandLen); //处理命令 }每个包中包含多个协议关键字,每个协议关键字用回车换行分开,因此我们需要调用文本分开函数,然后针对每条命令解析出关键字和值,具体代码在IncomingDataParser.DecodeProtocolText如下:
处理命令public bool DecodeProtocolText(string protocolText) { m_header = ""; m_names.Clear(); m_values.Clear(); int speIndex = protocolText.IndexOf(ProtocolKey.ReturnWrap); if (speIndex < 0) { return false; } else { string[] tmpNameValues = protocolText.Split(new string[] { ProtocolKey.ReturnWrap }, StringSplitOptions.RemoveEmptyEntries); if (tmpNameValues.Length < 2) //每次命令至少包括两行 return false; for (int i = 0; i < tmpNameValues.Length; i++) { string[] tmpStr = tmpNameValues[i].Split(new string[] { ProtocolKey.EqualSign }, StringSplitOptions.None); if (tmpStr.Length > 1) //存在等号 { if (tmpStr.Length > 2) //超过两个等号,返回失败 return false; if (tmpStr[0].Equals(ProtocolKey.Command, StringComparison.CurrentCultureIgnoreCase)) { m_command = tmpStr[1]; } else { m_names.Add(tmpStr[0].ToLower()); m_values.Add(tmpStr[1]); } } } return true; } }
解析出命令后,需要对每个命令进行处理,各个协议实现类从AsyncSocketInvokeElement.ProcessCommand继承,然后编写各自协议处理逻辑,如吞吐量的测试协议逻辑实现代码如下:
namespace SocketAsyncSvr { class ThroughputSocketProtocol : BaseSocketProtocol { public ThroughputSocketProtocol(AsyncSocketServer asyncSocketServer, AsyncSocketUserToken asyncSocketUserToken) : base(asyncSocketServer, asyncSocketUserToken) { m_socketFlag = "Throughput"; } public override void Close() { base.Close(); } public override bool ProcessCommand(byte[] buffer, int offset, int count) //处理分完包的数据,子类从这个方法继承 { ThroughputSocketCommand command = StrToCommand(m_incomingDataParser.Command); m_outgoingDataAssembler.Clear(); m_outgoingDataAssembler.AddResponse(); m_outgoingDataAssembler.AddCommand(m_incomingDataParser.Command); if (command == ThroughputSocketCommand.CyclePacket) return DoCyclePacket(buffer, offset, count); else { Program.Logger.Error("Unknow command: " + m_incomingDataParser.Command); return false; } } public ThroughputSocketCommand StrToCommand(string command) { if (command.Equals(ProtocolKey.CyclePacket, StringComparison.CurrentCultureIgnoreCase)) return ThroughputSocketCommand.CyclePacket; else return ThroughputSocketCommand.None; } public bool DoCyclePacket(byte[] buffer, int offset, int count) { int cycleCount = 0; if (m_incomingDataParser.GetValue(ProtocolKey.Count, ref cycleCount)) { m_outgoingDataAssembler.AddSuccess(); cycleCount = cycleCount + 1; m_outgoingDataAssembler.AddValue(ProtocolKey.Count, cycleCount); } else m_outgoingDataAssembler.AddFailure(ProtocolCode.ParameterError, ""); return DoSendResult(buffer, offset, count); } } }DEMO下载地址:http://download.csdn.net/detail/sqldebug_fan/7467745
免责声明:此代码只是为了演示C#完成端口编程,仅用于学习和研究,切勿用于商业用途。水平有限,C#也属于初学,错误在所难免,欢迎指正和指导。邮箱地址:fansheng_hx@163.com。