C#高性能大容量SOCKET并发(三):接收、发送

异步数据接收有可能收到的数据不是一个完整包,或者接收到的数据超过一个包的大小,因此我们需要把接收的数据进行缓存。异步发送我们也需要把每个发送的包加入到一个队列,然后通过队列逐个发送出去,如果每个都实时发送,有可能造成上一个数据包未发送完成,,这时再调用SendAsync会抛出异常,提示SocketAsyncEventArgs正在进行异步操作,因此我们需要建立接收缓存和发送缓存。

接收

通过Completed事件响应后调用AsyncSocketInvokeElement.ProcessReceive,在ProcessReceive中,我们把收到数据先写入一个缓存,然后进行分包,分包后压给包处理函数ProcessPacket,ProcessPacket函数然后调用ProcessCommand处理具体的命令,也是各个协议实现业务逻辑的地方,具体代码如下:

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;}}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); //处理命令}public virtual bool ProcessCommand(byte[] buffer, int offset, int count) //处理具体命令,子类从这个方法继承,buffer是收到的数据{return true;}发送

通过Completed事件响应后调用AsyncSocketInvokeElement.SendCompleted,在SendCompleted中我们需要在队列中清除已发送的包,并检测是否还有剩余需要发送的数据包,如果有,则继续发送,具体实现如下:

public virtual bool SendCompleted(){m_activeDT = DateTime.UtcNow;m_sendAsync = false;AsyncSendBufferManager asyncSendBufferManager = m_asyncSocketUserToken.SendBuffer;asyncSendBufferManager.ClearFirstPacket(); //清除已发送的包int offset = 0;int count = 0;if (asyncSendBufferManager.GetFirstPacket(ref offset, ref count)){m_sendAsync = true;return m_asyncSocketServer.SendAsyncEvent(m_asyncSocketUserToken.ConnectSocket, m_asyncSocketUserToken.SendEventArgs,asyncSendBufferManager.DynamicBufferManager.Buffer, offset, count);}elsereturn SendCallback();}//发送回调函数,用于连续下发数据public virtual bool SendCallback(){return true;}在AsyncSocketInvokeElement中提供函数给子类发送数据,业务逻辑是把当前数据包写入缓存,并检测当前是否正在发送包,如果正在发送,则等待回调,如果没有正在发送的数据包,则投递发送请求。public bool DoSendResult(){string commandText = m_outgoingDataAssembler.GetProtocolText();byte[] bufferUTF8 = Encoding.UTF8.GetBytes(commandText);int totalLength = sizeof(int) + bufferUTF8.Length; //获取总大小AsyncSendBufferManager asyncSendBufferManager = m_asyncSocketUserToken.SendBuffer;asyncSendBufferManager.StartPacket();asyncSendBufferManager.DynamicBufferManager.WriteInt(totalLength, false); //写入总大小asyncSendBufferManager.DynamicBufferManager.WriteInt(bufferUTF8.Length, false); //写入命令大小asyncSendBufferManager.DynamicBufferManager.WriteBuffer(bufferUTF8); //写入命令内容asyncSendBufferManager.EndPacket();bool result = true;if (!m_sendAsync){int packetOffset = 0;int packetCount = 0;if (asyncSendBufferManager.GetFirstPacket(ref packetOffset, ref packetCount)){m_sendAsync = true;result = m_asyncSocketServer.SendAsyncEvent(m_asyncSocketUserToken.ConnectSocket, m_asyncSocketUserToken.SendEventArgs,asyncSendBufferManager.DynamicBufferManager.Buffer, packetOffset, packetCount);}}return result;}public bool DoSendResult(byte[] buffer, int offset, int count){string commandText = m_outgoingDataAssembler.GetProtocolText();byte[] bufferUTF8 = Encoding.UTF8.GetBytes(commandText);int totalLength = sizeof(int) + bufferUTF8.Length + count; //获取总大小AsyncSendBufferManager asyncSendBufferManager = m_asyncSocketUserToken.SendBuffer;asyncSendBufferManager.StartPacket();asyncSendBufferManager.DynamicBufferManager.WriteInt(totalLength, false); //写入总大小asyncSendBufferManager.DynamicBufferManager.WriteInt(bufferUTF8.Length, false); //写入命令大小asyncSendBufferManager.DynamicBufferManager.WriteBuffer(bufferUTF8); //写入命令内容asyncSendBufferManager.DynamicBufferManager.WriteBuffer(buffer, offset, count); //写入二进制数据asyncSendBufferManager.EndPacket();bool result = true;if (!m_sendAsync){int packetOffset = 0;int packetCount = 0;if (asyncSendBufferManager.GetFirstPacket(ref packetOffset, ref packetCount)){m_sendAsync = true;result = m_asyncSocketServer.SendAsyncEvent(m_asyncSocketUserToken.ConnectSocket, m_asyncSocketUserToken.SendEventArgs,asyncSendBufferManager.DynamicBufferManager.Buffer, packetOffset, packetCount);}}return result;}public bool DoSendBuffer(byte[] buffer, int offset, int count){AsyncSendBufferManager asyncSendBufferManager = m_asyncSocketUserToken.SendBuffer;asyncSendBufferManager.StartPacket();asyncSendBufferManager.DynamicBufferManager.WriteBuffer(buffer, offset, count);asyncSendBufferManager.EndPacket();bool result = true;if (!m_sendAsync){int packetOffset = 0;int packetCount = 0;if (asyncSendBufferManager.GetFirstPacket(ref packetOffset, ref packetCount)){m_sendAsync = true;result = m_asyncSocketServer.SendAsyncEvent(m_asyncSocketUserToken.ConnectSocket, m_asyncSocketUserToken.SendEventArgs,asyncSendBufferManager.DynamicBufferManager.Buffer, packetOffset, packetCount);}}return result;}免责声明:此代码只是为了演示C#完成端口编程,仅用于学习和研究,切勿用于商业用途。水平有限,C#也属于初学,错误在所难免,欢迎指正和指导。邮箱地址:fansheng_hx@163.com。

大理的洱海形如人耳,风平浪静时,

C#高性能大容量SOCKET并发(三):接收、发送

相关文章:

你感兴趣的文章:

标签云: