C#中一个高性能异步socket封装库的实现思路分享
前言
创新互联公司是一家集网站建设,玉树企业网站建设,玉树品牌网站建设,网站定制,玉树网站建设报价,网络营销,网络优化,玉树网站推广为一体的创新建站企业,帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿,时刻以成就客户成长自我,坚持不断学习、思考、沉淀、净化自己,让我们为更多的企业打造出实用型网站。
socket是软件之间通讯最常用的一种方式。c#实现socket通讯有很多中方法,其中效率最高就是异步通讯。
异步通讯实际是利用windows完成端口(IOCP)来处理的,关于完成端口实现原理,大家可以参考网上文章。
我这里想强调的是采用完成端口机制的异步通讯是windows下效率最高的通讯方式,没有之一!
异步通讯比同步通讯处理要难很多,代码编写中会遇到许多“坑“。如果没有经验,很难完成。
我搜集了大量资料,完成了对异步socket的封装。此库已用稳定高效的运行几个月。
纵观网上的资料,我还没有遇到一个满意的封装库。许多文章把数据收发和协议处理杂糅在一块,代码非常难懂,也无法扩展。
在编写该库时,避免以上缺陷。将逻辑处理层次化,模块化!同时实现了高可用性与高性能。
为了使大家对通讯效率有初步了解,先看测试图。
主机配置情况
百兆带宽基本占满,cpu占用40%,我的电脑在空闲时,cpu占用大概20%,也就是说程序占用cpu 20%左右。
这个库是可扩展的,就是说即使10万个连接,收发同样的数据,cpu占用基本相同。
库的结构图
目标
即可作为服务端(监听)也可以作为客户端(主动连接)使用。
可以适应任何网络协议。收发的数据针对字节流或一个完整的包。对协议内容不做处理。
高可用性。将复杂的底层处理封装,对外接口非常友好。
高性能。最大限度优化处理。单机可支持数万连接,收发速度可达几百兆bit。
实现思路
网络处理逻辑可以分为以下几个部分:
网络监听 可以在多个端口实现监听。负责生成socket,生成的socket供后续处理。监听模块功能比较单一,如有必要,可对监听模块做进一步优化。
主动连接 可以异步或同步的连接对方。连接成功后,对socket的后续处理,与监听得到的socket完全一样。注:无论是监听得到的socket,还是连接得到的socket,后续处理完全一样。
Socket收发处理 每个socket对应一个收发实例,socket收发只针对字节流处理。收发时,做了优化。比如发送时,对数据做了沾包,提高发送性能;接收时,一次投递1K的数据。
组包处理 一般数据包都有包长度指示;比如 报头的前俩个字节表示长度,根据这个值就可以组成一个完整的包。
NetListener 监听
using System; using System.Net; using System.Net.Sockets; using System.Threading; namespace IocpCore { class NetListener { private Socket listenSocket; public ListenParam _listenParam { get; set; } public event ActionOnAcceptSocket; bool start; NetServer _netServer; public NetListener(NetServer netServer) { _netServer = netServer; } public int _acceptAsyncCount = 0; public bool StartListen() { try { start = true; IPEndPoint listenPoint = new IPEndPoint(IPAddress.Parse("0.0.0.0"), _listenParam._port); listenSocket = new Socket(listenPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp); listenSocket.Bind(listenPoint); listenSocket.Listen(200); Thread thread1 = new Thread(new ThreadStart(NetProcess)); thread1.Start(); StartAccept(); return true; } catch (Exception ex) { NetLogger.Log(string.Format("**监听异常!{0}", ex.Message)); return false; } } AutoResetEvent _acceptEvent = new AutoResetEvent(false); private void NetProcess() { while (start) { DealNewAccept(); _acceptEvent.WaitOne(1000 * 10); } } private void DealNewAccept() { try { if(_acceptAsyncCount <= 10) { StartAccept(); } while (true) { AsyncSocketClient client = _newSocketClientList.GetObj(); if (client == null) break; DealNewAccept(client); } } catch (Exception ex) { NetLogger.Log(string.Format("DealNewAccept 异常 {0}***{1}", ex.Message, ex.StackTrace)); } } private void DealNewAccept(AsyncSocketClient client) { client.SendBufferByteCount = _netServer.SendBufferBytePerClient; OnAcceptSocket?.Invoke(_listenParam, client); } private void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs acceptEventArgs) { try { Interlocked.Decrement(ref _acceptAsyncCount); _acceptEvent.Set(); acceptEventArgs.Completed -= AcceptEventArg_Completed; ProcessAccept(acceptEventArgs); } catch (Exception ex) { NetLogger.Log(string.Format("AcceptEventArg_Completed {0}***{1}", ex.Message, ex.StackTrace)); } } public bool StartAccept() { SocketAsyncEventArgs acceptEventArgs = new SocketAsyncEventArgs(); acceptEventArgs.Completed += AcceptEventArg_Completed; bool willRaiseEvent = listenSocket.AcceptAsync(acceptEventArgs); Interlocked.Increment(ref _acceptAsyncCount); if (!willRaiseEvent) { Interlocked.Decrement(ref _acceptAsyncCount); _acceptEvent.Set(); acceptEventArgs.Completed -= AcceptEventArg_Completed; ProcessAccept(acceptEventArgs); } return true; } ObjectPool _newSocketClientList = new ObjectPool (); private void ProcessAccept(SocketAsyncEventArgs acceptEventArgs) { try { using (acceptEventArgs) { if (acceptEventArgs.AcceptSocket != null) { AsyncSocketClient client = new AsyncSocketClient(acceptEventArgs.AcceptSocket); client.CreateClientInfo(this); _newSocketClientList.PutObj(client); _acceptEvent.Set(); } } } catch (Exception ex) { NetLogger.Log(string.Format("ProcessAccept {0}***{1}", ex.Message, ex.StackTrace)); } } } }
NetConnectManage连接处理
using System; using System.Net; using System.Net.Sockets; namespace IocpCore { class NetConnectManage { public event ActionOnSocketConnectEvent; public bool ConnectAsyn(string peerIp, int peerPort, object tag) { try { Socket socket = new Socket(SocketType.Stream, ProtocolType.Tcp); SocketAsyncEventArgs socketEventArgs = new SocketAsyncEventArgs(); socketEventArgs.RemoteEndPoint = new IPEndPoint(IPAddress.Parse(peerIp), peerPort); socketEventArgs.Completed += SocketConnect_Completed; SocketClientInfo clientInfo = new SocketClientInfo(); socketEventArgs.UserToken = clientInfo; clientInfo.PeerIp = peerIp; clientInfo.PeerPort = peerPort; clientInfo.Tag = tag; bool willRaiseEvent = socket.ConnectAsync(socketEventArgs); if (!willRaiseEvent) { ProcessConnect(socketEventArgs); socketEventArgs.Completed -= SocketConnect_Completed; socketEventArgs.Dispose(); } return true; } catch (Exception ex) { NetLogger.Log("ConnectAsyn",ex); return false; } } private void SocketConnect_Completed(object sender, SocketAsyncEventArgs socketEventArgs) { ProcessConnect(socketEventArgs); socketEventArgs.Completed -= SocketConnect_Completed; socketEventArgs.Dispose(); } private void ProcessConnect(SocketAsyncEventArgs socketEventArgs) { SocketClientInfo clientInfo = socketEventArgs.UserToken as SocketClientInfo; if (socketEventArgs.SocketError == SocketError.Success) { DealConnectSocket(socketEventArgs.ConnectSocket, clientInfo); } else { SocketEventParam socketParam = new SocketEventParam(EN_SocketEvent.connect, null); socketParam.ClientInfo = clientInfo; OnSocketConnectEvent?.Invoke(socketParam, null); } } void DealConnectSocket(Socket socket, SocketClientInfo clientInfo) { clientInfo.SetClientInfo(socket); AsyncSocketClient client = new AsyncSocketClient(socket); client.SetClientInfo(clientInfo); //触发事件 SocketEventParam socketParam = new SocketEventParam(EN_SocketEvent.connect, socket); socketParam.ClientInfo = clientInfo; OnSocketConnectEvent?.Invoke(socketParam, client); } public bool Connect(string peerIp, int peerPort, object tag, out Socket socket) { socket = null; try { Socket socketTmp = new Socket(SocketType.Stream, ProtocolType.Tcp); SocketClientInfo clientInfo = new SocketClientInfo(); clientInfo.PeerIp = peerIp; clientInfo.PeerPort = peerPort; clientInfo.Tag = tag; EndPoint remoteEP = new IPEndPoint(IPAddress.Parse(peerIp), peerPort); socketTmp.Connect(remoteEP); if (!socketTmp.Connected) return false; DealConnectSocket(socketTmp, clientInfo); socket = socketTmp; return true; } catch (Exception ex) { NetLogger.Log(string.Format("连接对方:({0}:{1})出错!", peerIp, peerPort), ex); return false; } } } }
AsyncSocketClient socket收发处理
using System; using System.Collections.Generic; using System.Diagnostics; using System.Net; using System.Net.Sockets; namespace IocpCore { public class AsyncSocketClient { public static int IocpReadLen = 1024; public readonly Socket ConnectSocket; protected SocketAsyncEventArgs m_receiveEventArgs; public SocketAsyncEventArgs ReceiveEventArgs { get { return m_receiveEventArgs; } set { m_receiveEventArgs = value; } } protected byte[] m_asyncReceiveBuffer; protected SocketAsyncEventArgs m_sendEventArgs; public SocketAsyncEventArgs SendEventArgs { get { return m_sendEventArgs; } set { m_sendEventArgs = value; } } protected byte[] m_asyncSendBuffer; public event ActionOnReadData; public event Action OnSendData; public event Action OnSocketClose; static object releaseLock = new object(); public static int createCount = 0; public static int releaseCount = 0; ~AsyncSocketClient() { lock (releaseLock) { releaseCount++; } } public AsyncSocketClient(Socket socket) { lock (releaseLock) { createCount++; } ConnectSocket = socket; m_receiveEventArgs = new SocketAsyncEventArgs(); m_asyncReceiveBuffer = new byte[IocpReadLen]; m_receiveEventArgs.AcceptSocket = ConnectSocket; m_receiveEventArgs.Completed += ReceiveEventArgs_Completed; m_sendEventArgs = new SocketAsyncEventArgs(); m_asyncSendBuffer = new byte[IocpReadLen * 2]; m_sendEventArgs.AcceptSocket = ConnectSocket; m_sendEventArgs.Completed += SendEventArgs_Completed; } SocketClientInfo _clientInfo; public SocketClientInfo ClientInfo { get { return _clientInfo; } } internal void CreateClientInfo(NetListener netListener) { _clientInfo = new SocketClientInfo(); try { _clientInfo.Tag = netListener._listenParam._tag; IPEndPoint ip = ConnectSocket.LocalEndPoint as IPEndPoint; Debug.Assert(netListener._listenParam._port == ip.Port); _clientInfo.LocalIp = ip.Address.ToString(); _clientInfo.LocalPort = netListener._listenParam._port; ip = ConnectSocket.RemoteEndPoint as IPEndPoint; _clientInfo.PeerIp = ip.Address.ToString(); _clientInfo.PeerPort = ip.Port; } catch (Exception ex) { NetLogger.Log("CreateClientInfo", ex); } } internal void SetClientInfo(SocketClientInfo clientInfo) { _clientInfo = clientInfo; } #region read process bool _inReadPending = false; public EN_SocketReadResult ReadNextData() { lock (this) { if (_socketError) return EN_SocketReadResult.ReadError; if (_inReadPending) return EN_SocketReadResult.InAsyn; if(!ConnectSocket.Connected) { OnReadError(); return EN_SocketReadResult.ReadError; } try { m_receiveEventArgs.SetBuffer(m_asyncReceiveBuffer, 0, m_asyncReceiveBuffer.Length); _inReadPending = true; bool willRaiseEvent = ConnectSocket.ReceiveAsync(ReceiveEventArgs); //投递接收请求 if (!willRaiseEvent) { _inReadPending = false; ProcessReceive(); if (_socketError) { OnReadError(); return EN_SocketReadResult.ReadError; } return EN_SocketReadResult.HaveRead; } else { return EN_SocketReadResult.InAsyn; } } catch (Exception ex) { NetLogger.Log("ReadNextData", ex); _inReadPending = false; OnReadError(); return EN_SocketReadResult.ReadError; } } } private void ProcessReceive() { if (ReceiveEventArgs.BytesTransferred > 0 && ReceiveEventArgs.SocketError == SocketError.Success) { int offset = ReceiveEventArgs.Offset; int count = ReceiveEventArgs.BytesTransferred; byte[] readData = new byte[count]; Array.Copy(m_asyncReceiveBuffer, offset, readData, 0, count); _inReadPending = false; if (!_socketError) OnReadData?.Invoke(this, readData); } else { _inReadPending = false; OnReadError(); } } private void ReceiveEventArgs_Completed(object sender, SocketAsyncEventArgs e) { lock (this) { _inReadPending = false; ProcessReceive(); if (_socketError) { OnReadError(); } } } bool _socketError = false; private void OnReadError() { lock (this) { if (_socketError == false) { _socketError = true; OnSocketClose?.Invoke(this); } CloseClient(); } } #endregion #region send process int _sendBufferByteCount = 102400; public int SendBufferByteCount { get { return _sendBufferByteCount; } set { if (value < 1024) { _sendBufferByteCount = 1024; } else { _sendBufferByteCount = value; } } } SendBufferPool _sendDataPool = new SendBufferPool(); internal EN_SendDataResult PutSendData(byte[] data) { if (_socketError) return EN_SendDataResult.no_client; if (_sendDataPool._bufferByteCount >= _sendBufferByteCount) { return EN_SendDataResult.buffer_overflow; } if (data.Length <= IocpReadLen) { _sendDataPool.PutObj(data); } else { List dataItems = SplitData(data, IocpReadLen); foreach (byte[] item in dataItems) { _sendDataPool.PutObj(item); } } return EN_SendDataResult.ok; } bool _inSendPending = false; public EN_SocketSendResult SendNextData() { lock (this) { if (_socketError) { return EN_SocketSendResult.SendError; } if (_inSendPending) { return EN_SocketSendResult.InAsyn; } int sendByteCount = GetSendData(); if (sendByteCount == 0) { return EN_SocketSendResult.NoSendData; } //防止抛出异常,否则影响性能 if (!ConnectSocket.Connected) { OnSendError(); return EN_SocketSendResult.SendError; } try { m_sendEventArgs.SetBuffer(m_asyncSendBuffer, 0, sendByteCount); _inSendPending = true; bool willRaiseEvent = ConnectSocket.SendAsync(m_sendEventArgs); if (!willRaiseEvent) { _inSendPending = false; ProcessSend(m_sendEventArgs); if (_socketError) { OnSendError(); return EN_SocketSendResult.SendError; } else { OnSendData?.Invoke(this, sendByteCount); //继续发下一条 return EN_SocketSendResult.HaveSend; } } else { return EN_SocketSendResult.InAsyn; } } catch (Exception ex) { NetLogger.Log("SendNextData", ex); _inSendPending = false; OnSendError(); return EN_SocketSendResult.SendError; } } } private void SendEventArgs_Completed(object sender, SocketAsyncEventArgs sendEventArgs) { lock (this) { try { _inSendPending = false; ProcessSend(m_sendEventArgs); int sendCount = 0; if (sendEventArgs.SocketError == SocketError.Success) { sendCount = sendEventArgs.BytesTransferred; } OnSendData?.Invoke(this, sendCount); if (_socketError) { OnSendError(); } } catch (Exception ex) { NetLogger.Log("SendEventArgs_Completed", ex); } } } private bool ProcessSend(SocketAsyncEventArgs sendEventArgs) { if (sendEventArgs.SocketError == SocketError.Success) { return true; } else { OnSendError(); return false; } } private int GetSendData() { int dataLen = 0; while (true) { byte[] data = _sendDataPool.GetObj(); if (data == null) return dataLen; Array.Copy(data, 0, m_asyncSendBuffer, dataLen, data.Length); dataLen += data.Length; if (dataLen > IocpReadLen) break; } return dataLen; } private void OnSendError() { lock (this) { if (_socketError == false) { _socketError = true; OnSocketClose?.Invoke(this); } CloseClient(); } } #endregion internal void CloseSocket() { try { ConnectSocket.Close(); } catch (Exception ex) { NetLogger.Log("CloseSocket", ex); } } static object socketCloseLock = new object(); public static int closeSendCount = 0; public static int closeReadCount = 0; bool _disposeSend = false; void CloseSend() { if (!_disposeSend && !_inSendPending) { lock (socketCloseLock) closeSendCount++; _disposeSend = true; m_sendEventArgs.SetBuffer(null, 0, 0); m_sendEventArgs.Completed -= SendEventArgs_Completed; m_sendEventArgs.Dispose(); } } bool _disposeRead = false; void CloseRead() { if (!_disposeRead && !_inReadPending) { lock (socketCloseLock) closeReadCount++; _disposeRead = true; m_receiveEventArgs.SetBuffer(null, 0, 0); m_receiveEventArgs.Completed -= ReceiveEventArgs_Completed; m_receiveEventArgs.Dispose(); } } private void CloseClient() { try { CloseSend(); CloseRead(); ConnectSocket.Close(); } catch (Exception ex) { NetLogger.Log("CloseClient", ex); } } //发送缓冲大小 private List SplitData(byte[] data, int maxLen) { List items = new List (); int start = 0; while (true) { int itemLen = Math.Min(maxLen, data.Length - start); if (itemLen == 0) break; byte[] item = new byte[itemLen]; Array.Copy(data, start, item, 0, itemLen); items.Add(item); start += itemLen; } return items; } } public enum EN_SocketReadResult { InAsyn, HaveRead, ReadError } public enum EN_SocketSendResult { InAsyn, HaveSend, NoSendData, SendError } class SendBufferPool { ObjectPool _bufferPool = new ObjectPool (); public Int64 _bufferByteCount = 0; public bool PutObj(byte[] obj) { if (_bufferPool.PutObj(obj)) { lock (this) { _bufferByteCount += obj.Length; } return true; } else { return false; } } public byte[] GetObj() { byte[] result = _bufferPool.GetObj(); if (result != null) { lock (this) { _bufferByteCount -= result.Length; } } return result; } } }
NetServer 聚合其他类
using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Net.Sockets; using System.Threading; namespace IocpCore { public class NetServer { public ActionOnSocketPacketEvent; //每个连接发送缓冲大小 public int SendBufferBytePerClient { get; set; } = 1024 * 100; bool _serverStart = false; List _listListener = new List (); //负责对收到的字节流 组成完成的包 ClientPacketManage _clientPacketManage; public Int64 SendByteCount { get; set; } public Int64 ReadByteCount { get; set; } List _listListenPort = new List (); public void AddListenPort(int port, object tag) { _listListenPort.Add(new ListenParam(port, tag)); } /// /// /// /// 监听失败的端口 ///public bool StartListen(out List listenFault) { _serverStart = true; _clientPacketManage = new ClientPacketManage(this); _clientPacketManage.OnSocketPacketEvent += PutClientPacket; _netConnectManage.OnSocketConnectEvent += SocketConnectEvent; _listListener.Clear(); Thread thread1 = new Thread(new ThreadStart(NetPacketProcess)); thread1.Start(); Thread thread2 = new Thread(new ThreadStart(NetSendProcess)); thread2.Start(); Thread thread3 = new Thread(new ThreadStart(NetReadProcess)); thread3.Start(); listenFault = new List (); foreach (ListenParam param in _listListenPort) { NetListener listener = new NetListener(this); listener._listenParam = param; listener.OnAcceptSocket += Listener_OnAcceptSocket; if (!listener.StartListen()) { listenFault.Add(param._port); } else { _listListener.Add(listener); NetLogger.Log(string.Format("监听成功!端口:{0}", param._port)); } } return listenFault.Count == 0; } public void PutClientPacket(SocketEventParam param) { OnSocketPacketEvent?.Invoke(param); } //获取包的最小长度 int _packetMinLen; int _packetMaxLen; public int PacketMinLen { get { return _packetMinLen; } } public int PacketMaxLen { get { return _packetMaxLen; } } /// /// 设置包的最小和最大长度 /// 当minLen=0时,认为是接收字节流 /// /// /// public void SetPacketParam(int minLen, int maxLen) { Debug.Assert(minLen >= 0); Debug.Assert(maxLen > minLen); _packetMinLen = minLen; _packetMaxLen = maxLen; } //获取包的总长度 public delegate int delegate_GetPacketTotalLen(byte[] data, int offset); public delegate_GetPacketTotalLen GetPacketTotalLen_Callback; ObjectPoolWithEvent_socketEventPool = new ObjectPoolWithEvent (); private void NetPacketProcess() { while (_serverStart) { try { DealEventPool(); } catch (Exception ex) { NetLogger.Log(string.Format("DealEventPool 异常 {0}***{1}", ex.Message, ex.StackTrace)); } _socketEventPool.WaitOne(1000); } } Dictionary _clientGroup = new Dictionary (); public int ClientCount { get { lock (_clientGroup) { return _clientGroup.Count; } } } public List ClientList { get { lock (_clientGroup) { return _clientGroup.Keys.ToList(); } } } private void DealEventPool() { while (true) { SocketEventParam param = _socketEventPool.GetObj(); if (param == null) return; if (param.SocketEvent == EN_SocketEvent.close) { lock (_clientGroup) { _clientGroup.Remove(param.Socket); } } if (_packetMinLen == 0)//字节流处理 { OnSocketPacketEvent?.Invoke(param); } else { //组成一个完整的包 逻辑 _clientPacketManage.PutSocketParam(param); } } } private void SocketConnectEvent(SocketEventParam param, AsyncSocketClient client) { try { if (param.Socket == null || client == null) //连接失败 { } else { lock (_clientGroup) { bool remove = _clientGroup.Remove(client.ConnectSocket); Debug.Assert(!remove); _clientGroup.Add(client.ConnectSocket, client); } client.OnSocketClose += Client_OnSocketClose; client.OnReadData += Client_OnReadData; client.OnSendData += Client_OnSendData; _listReadEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.read)); } _socketEventPool.PutObj(param); } catch (Exception ex) { NetLogger.Log(string.Format("SocketConnectEvent 异常 {0}***{1}", ex.Message, ex.StackTrace)); } } internal void OnRcvPacketLenError(Socket socket, byte[] buffer, int offset, int packetLen) { try { lock (_clientGroup) { if (!_clientGroup.ContainsKey(socket)) { Debug.Assert(false); return; } NetLogger.Log(string.Format("报长度异常!包长:{0}", packetLen)); AsyncSocketClient client = _clientGroup[socket]; client.CloseSocket(); } } catch (Exception ex) { NetLogger.Log(string.Format("OnRcvPacketLenError 异常 {0}***{1}", ex.Message, ex.StackTrace)); } } #region listen port private void Listener_OnAcceptSocket(ListenParam listenPatam, AsyncSocketClient client) { try { lock (_clientGroup) { bool remove = _clientGroup.Remove(client.ConnectSocket); Debug.Assert(!remove); _clientGroup.Add(client.ConnectSocket, client); } client.OnSocketClose += Client_OnSocketClose; client.OnReadData += Client_OnReadData; client.OnSendData += Client_OnSendData; _listReadEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.read)); SocketEventParam param = new SocketEventParam(EN_SocketEvent.accept, client.ConnectSocket); param.ClientInfo = client.ClientInfo; _socketEventPool.PutObj(param); } catch (Exception ex) { NetLogger.Log(string.Format("Listener_OnAcceptSocket 异常 {0}***{1}", ex.Message, ex.StackTrace)); } } ObjectPoolWithEvent _listSendEvent = new ObjectPoolWithEvent (); private void NetSendProcess() { while (true) { DealSendEvent(); _listSendEvent.WaitOne(1000); } } ObjectPoolWithEvent _listReadEvent = new ObjectPoolWithEvent (); private void NetReadProcess() { while (true) { DealReadEvent(); _listReadEvent.WaitOne(1000); } } private void DealSendEvent() { while (true) { SocketEventDeal item = _listSendEvent.GetObj(); if (item == null) break; switch (item.SocketEvent) { case EN_SocketDealEvent.send: { while (true) { EN_SocketSendResult result = item.Client.SendNextData(); if (result == EN_SocketSendResult.HaveSend) continue; else break; } } break; case EN_SocketDealEvent.read: { Debug.Assert(false); } break; } } } private void DealReadEvent() { while (true) { SocketEventDeal item = _listReadEvent.GetObj(); if (item == null) break; switch (item.SocketEvent) { case EN_SocketDealEvent.read: { while (true) { EN_SocketReadResult result = item.Client.ReadNextData(); if (result == EN_SocketReadResult.HaveRead) continue; else break; } } break; case EN_SocketDealEvent.send: { Debug.Assert(false); } break; } } } private void Client_OnReadData(AsyncSocketClient client, byte[] readData) { //读下一条 _listReadEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.read)); try { SocketEventParam param = new SocketEventParam(EN_SocketEvent.read, client.ConnectSocket); param.ClientInfo = client.ClientInfo; param.Data = readData; _socketEventPool.PutObj(param); lock (this) { ReadByteCount += readData.Length; } } catch (Exception ex) { NetLogger.Log(string.Format("Client_OnReadData 异常 {0}***{1}", ex.Message, ex.StackTrace)); } } #endregion private void Client_OnSendData(AsyncSocketClient client, int sendCount) { //发送下一条 _listSendEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.send)); lock (this) { SendByteCount += sendCount; } } private void Client_OnSocketClose(AsyncSocketClient client) { try { SocketEventParam param = new SocketEventParam(EN_SocketEvent.close, client.ConnectSocket); param.ClientInfo = client.ClientInfo; _socketEventPool.PutObj(param); } catch (Exception ex) { NetLogger.Log(string.Format("Client_OnSocketClose 异常 {0}***{1}", ex.Message, ex.StackTrace)); } } /// /// 放到发送缓冲 /// /// /// ///public EN_SendDataResult SendData(Socket socket, byte[] data) { if (socket == null) return EN_SendDataResult.no_client; lock (_clientGroup) { if (!_clientGroup.ContainsKey(socket)) return EN_SendDataResult.no_client; AsyncSocketClient client = _clientGroup[socket]; EN_SendDataResult result = client.PutSendData(data); if (result == EN_SendDataResult.ok) { //发送下一条 _listSendEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.send)); } return result; } } /// /// 设置某个连接的发送缓冲大小 /// /// /// ///public bool SetClientSendBuffer(Socket socket, int byteCount) { lock (_clientGroup) { if (!_clientGroup.ContainsKey(socket)) return false; AsyncSocketClient client = _clientGroup[socket]; client.SendBufferByteCount = byteCount; return true; } } #region connect process NetConnectManage _netConnectManage = new NetConnectManage(); /// /// 异步连接一个客户端 /// /// /// /// ///public bool ConnectAsyn(string peerIp, int peerPort, object tag) { return _netConnectManage.ConnectAsyn(peerIp, peerPort, tag); } /// /// 同步连接一个客户端 /// /// /// /// /// ///public bool Connect(string peerIp, int peerPort, object tag, out Socket socket) { return _netConnectManage.Connect(peerIp, peerPort, tag, out socket); } #endregion } enum EN_SocketDealEvent { read, send, } class SocketEventDeal { public AsyncSocketClient Client { get; set; } public EN_SocketDealEvent SocketEvent { get; set; } public SocketEventDeal(AsyncSocketClient client, EN_SocketDealEvent socketEvent) { Client = client; SocketEvent = socketEvent; } } }
库的使用
使用起来非常简单,示例如下
using IocpCore; using System; using System.Collections.Generic; using System.Linq; using System.Net.Sockets; using System.Text; using System.Threading.Tasks; using System.Windows; namespace WarningClient { public class SocketServer { public ActionOnSocketEvent; public Int64 SendByteCount { get { if (_netServer == null) return 0; return _netServer.SendByteCount; } } public Int64 ReadByteCount { get { if (_netServer == null) return 0; return _netServer.ReadByteCount; } } NetServer _netServer; EN_PacketType _packetType = EN_PacketType.byteStream; public void SetPacktType(EN_PacketType packetType) { _packetType = packetType; if (_netServer == null) return; if (packetType == EN_PacketType.byteStream) { _netServer.SetPacketParam(0, 1024); } else { _netServer.SetPacketParam(9, 1024); } } public bool Init(List listenPort) { NetLogger.OnLogEvent += NetLogger_OnLogEvent; _netServer = new NetServer(); SetPacktType(_packetType); _netServer.GetPacketTotalLen_Callback += GetPacketTotalLen; _netServer.OnSocketPacketEvent += SocketPacketDeal; foreach (int n in listenPort) { _netServer.AddListenPort(n, n); } List listenFault; bool start = _netServer.StartListen(out listenFault); return start; } int GetPacketTotalLen(byte[] data, int offset) { if (MainWindow._packetType == EN_PacketType.znss) return GetPacketZnss(data, offset); else return GetPacketAnzhiyuan(data, offset); } int GetPacketAnzhiyuan(byte[] data, int offset) { int n = data[offset + 5] + 6; return n; } int GetPacketZnss(byte[] data, int offset) { int packetLen = (int)(data[4]) + 5; return packetLen; } public bool ConnectAsyn(string peerIp, int peerPort, object tag) { return _netServer.ConnectAsyn(peerIp, peerPort, tag); } public bool Connect(string peerIp, int peerPort, object tag, out Socket socket) { return _netServer.Connect(peerIp, peerPort, tag, out socket); } private void NetLogger_OnLogEvent(string message) { AppLog.Log(message); } Dictionary _clientGroup = new Dictionary (); public int ClientCount { get { lock (_clientGroup) { return _clientGroup.Count; } } } public List ClientList { get { if (_netServer != null) return _netServer.ClientList; return new List (); } } void AddClient(SocketEventParam socketParam) { lock (_clientGroup) { _clientGroup.Remove(socketParam.Socket); _clientGroup.Add(socketParam.Socket, socketParam); } } void RemoveClient(SocketEventParam socketParam) { lock (_clientGroup) { _clientGroup.Remove(socketParam.Socket); } } ObjectPool _readDataPool = new ObjectPool (); public ObjectPool ReadDataPool { get { return _readDataPool; } } private void SocketPacketDeal(SocketEventParam socketParam) { OnSocketEvent?.Invoke(socketParam); if (socketParam.SocketEvent == EN_SocketEvent.read) { if (MainWindow._isShowReadPacket) _readDataPool.PutObj(socketParam); } else if (socketParam.SocketEvent == EN_SocketEvent.accept) { AddClient(socketParam); string peerIp = socketParam.ClientInfo.PeerIpPort; AppLog.Log(string.Format("客户端链接!本地端口:{0},对端:{1}", socketParam.ClientInfo.LocalPort, peerIp)); } else if (socketParam.SocketEvent == EN_SocketEvent.connect) { string peerIp = socketParam.ClientInfo.PeerIpPort; if (socketParam.Socket != null) { AddClient(socketParam); AppLog.Log(string.Format("连接对端成功!本地端口:{0},对端:{1}", socketParam.ClientInfo.LocalPort, peerIp)); } else { AppLog.Log(string.Format("连接对端失败!本地端口:{0},对端:{1}", socketParam.ClientInfo.LocalPort, peerIp)); } } else if (socketParam.SocketEvent == EN_SocketEvent.close) { MainWindow.MainWnd.OnSocketDisconnect(socketParam.Socket); RemoveClient(socketParam); string peerIp = socketParam.ClientInfo.PeerIpPort; AppLog.Log(string.Format("客户端断开!本地端口:{0},对端:{1},", socketParam.ClientInfo.LocalPort, peerIp)); } } public EN_SendDataResult SendData(Socket socket, byte[] data) { if(socket == null) { MessageBox.Show("还没连接!"); return EN_SendDataResult.no_client; } return _netServer.SendData(socket, data); } internal void SendToAll(byte[] data) { lock (_clientGroup) { foreach (Socket socket in _clientGroup.Keys) { SendData(socket, data); } } } } }
以上这篇C#中一个高性能异步socket封装库的实现思路分享就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持创新互联。
文章题目:C#中一个高性能异步socket封装库的实现思路分享
标题URL:http://myzitong.com/article/gosdoi.html