logo
搜索
我的图书馆

发文章

发文工具

其他工具

留言交流
搜索
分享

C# 一个高性能异步socket封装库的实现思路

 眼镜蛇图书馆zh 2019-08-31

转：https://www.cnblogs.com/yuanchenhui/p/asyn_scoket.html

前言

socket是软件之间通讯最常用的一种方式。c#实现socket通讯有很多中方法，其中效率最高就是异步通讯。

异步通讯实际是利用windows完成端口（IOCP）来处理的，关于完成端口实现原理，大家可以参考网上文章。

我这里想强调的是采用完成端口机制的异步通讯是windows下效率最高的通讯方式，没有之一！

 

异步通讯比同步通讯处理要难很多，代码编写中会遇到许多“坑“。如果没有经验，很难完成。

我搜集了大量资料，完成了对异步socket的封装。此库已用稳定高效的运行几个月。

 

纵观网上的资料，我还没有遇到一个满意的封装库。许多文章把数据收发和协议处理杂糅在一块，代码非常难懂，也无法扩展。

在编写该库时，避免以上缺陷。将逻辑处理层次化，模块化！同时实现了高可用性与高性能。

 

为了使大家对通讯效率有初步了解，先看测试图。

客户端和服务端都是本机测试，最大连接数为64422，套接字已耗尽！

主机配置情况

百兆带宽基本占满，cpu占用40%，我的电脑在空闲时，cpu占用大概20%，也就是说程序占用cpu 20%左右。

这个库是可扩展的，就是说即使10万个连接，收发同样的数据，cpu占用基本相同。

 

库的结构图 

 

目标

  1. 即可作为服务端（监听）也可以作为客户端（主动连接）使用。
  2. 可以适应任何网络协议。收发的数据针对字节流或一个完整的包。对协议内容不做处理。
  3. 高可用性。将复杂的底层处理封装，对外接口非常友好。
  4. 高性能。最大限度优化处理。单机可支持数万连接，收发速度可达几百兆bit。

实现思路

网络处理逻辑可以分为以下几个部分：

  1. 网络监听   可以在多个端口实现监听。负责生成socket，生成的socket供后续处理。监听模块功能比较单一，如有必要，可对监听模块做进一步优化。
  2. 主动连接  可以异步或同步的连接对方。连接成功后，对socket的后续处理，与监听得到的socket完全一样。注：无论是监听得到的socket，还是连接得到的socket，后续处理完全一样。
  3. Socket收发处理   每个socket对应一个收发实例，socket收发只针对字节流处理。收发时，做了优化。比如发送时，对数据做了沾包，提高发送性能；接收时，一次投递1K的数据。
  4. 组包处理   一般数据包都有包长度指示；比如 报头的前俩个字节表示长度，根据这个值就可以组成一个完整的包。

 NetListener 监听

  1. using System;
  2. using System.Net;
  3. using System.Net.Sockets;
  4. using System.Threading;

  6. namespace IocpCore
  7. {
  8.     class NetListener
  9.     {
  10.         private Socket listenSocket;
  11.         public ListenParam _listenParam { get; set; }
  12.         public event Action<ListenParam, AsyncSocketClient> OnAcceptSocket;

  14.         bool start;

  16.         NetServer _netServer;
  17.         public NetListener(NetServer netServer)
  18.         {
  19.             _netServer = netServer;
  20.         }

  22.         public int _acceptAsyncCount = 0;
  23.         public bool StartListen()
  24.         {
  25.             try
  26.             {
  27.                 start = true;
  28.                 IPEndPoint listenPoint = new IPEndPoint(IPAddress.Parse("0.0.0.0"), _listenParam._port);
  29.                 listenSocket = new Socket(listenPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
  30.                 listenSocket.Bind(listenPoint);
  31.                 listenSocket.Listen(200);

  33.                 Thread thread1 = new Thread(new ThreadStart(NetProcess));
  34.                 thread1.Start();

  36.                 StartAccept();
  37.                 return true;
  38.             }
  39.             catch (Exception ex)
  40.             {
  41.                 NetLogger.Log(string.Format("**监听异常!{0}", ex.Message));
  42.                 return false;
  43.             }
  44.         }

  46.         AutoResetEvent _acceptEvent = new AutoResetEvent(false);
  47.         private void NetProcess()
  48.         {
  49.             while (start)
  50.             {
  51.                 DealNewAccept();
  52.                 _acceptEvent.WaitOne(1000 * 10);
  53.             }
  54.         }

  56.         private void DealNewAccept()
  57.         {
  58.             try
  59.             {
  60.                 if(_acceptAsyncCount <= 10)
  61.                 {
  62.                     StartAccept();
  63.                 }

  65.                 while (true)
  66.                 {
  67.                     AsyncSocketClient client = _newSocketClientList.GetObj();
  68.                     if (client == null)
  69.                         break;

  71.                     DealNewAccept(client);
  72.                 }
  73.             }
  74.             catch (Exception ex)
  75.             {
  76.                 NetLogger.Log(string.Format("DealNewAccept 异常 {0}***{1}", ex.Message, ex.StackTrace));
  77.             }
  78.         }

  80.         private void DealNewAccept(AsyncSocketClient client)
  81.         {
  82.             client.SendBufferByteCount = _netServer.SendBufferBytePerClient;
  83.            OnAcceptSocket?.Invoke(_listenParam, client);
  84.         }

  86.         private void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs acceptEventArgs)
  87.         {
  88.             try
  89.             {
  90.                 Interlocked.Decrement(ref _acceptAsyncCount);
  91.                 _acceptEvent.Set();
  92.                 acceptEventArgs.Completed -= AcceptEventArg_Completed;
  93.                 ProcessAccept(acceptEventArgs);
  94.             }
  95.             catch (Exception ex)
  96.             {
  97.                 NetLogger.Log(string.Format("AcceptEventArg_Completed {0}***{1}", ex.Message, ex.StackTrace));
  98.             }
  99.         }

  101.         public bool StartAccept()
  102.         {
  103.             SocketAsyncEventArgs acceptEventArgs = new SocketAsyncEventArgs();
  104.             acceptEventArgs.Completed += AcceptEventArg_Completed;

  106.             bool willRaiseEvent = listenSocket.AcceptAsync(acceptEventArgs);
  107.             Interlocked.Increment(ref _acceptAsyncCount);

  109.             if (!willRaiseEvent)
  110.             {
  111.                 Interlocked.Decrement(ref _acceptAsyncCount);
  112.                 _acceptEvent.Set();
  113.                 acceptEventArgs.Completed -= AcceptEventArg_Completed;
  114.                 ProcessAccept(acceptEventArgs);
  115.             }
  116.             return true;
  117.         }

  119.         ObjectPool<AsyncSocketClient> _newSocketClientList = new ObjectPool<AsyncSocketClient>();
  120.         private void ProcessAccept(SocketAsyncEventArgs acceptEventArgs)
  121.         {
  122.             try
  123.             {
  124.                 using (acceptEventArgs)
  125.                 {
  126.                     if (acceptEventArgs.AcceptSocket != null)
  127.                     {
  128.                         AsyncSocketClient client = new AsyncSocketClient(acceptEventArgs.AcceptSocket);
  129.                         client.CreateClientInfo(this);

  131.                         _newSocketClientList.PutObj(client);
  132.                         _acceptEvent.Set();
  133.                     }
  134.                 }
  135.             }
  136.             catch (Exception ex)
  137.             {
  138.                 NetLogger.Log(string.Format("ProcessAccept {0}***{1}", ex.Message, ex.StackTrace));
  139.             }
  140.         }
  141.     }
  142. }

NetConnectManage连接处理

  1. using System;
  2. using System.Net;
  3. using System.Net.Sockets;

  5. namespace IocpCore
  6. {
  7.     class NetConnectManage
  8.     {
  9.         public event Action<SocketEventParam, AsyncSocketClient> OnSocketConnectEvent;

  11.         public bool ConnectAsyn(string peerIp, int peerPort, object tag)
  12.         {
  13.             try
  14.             {
  15.                 Socket socket = new Socket(SocketType.Stream, ProtocolType.Tcp);
  16.                 SocketAsyncEventArgs socketEventArgs = new SocketAsyncEventArgs();
  17.                 socketEventArgs.RemoteEndPoint = new IPEndPoint(IPAddress.Parse(peerIp), peerPort);
  18.                 socketEventArgs.Completed += SocketConnect_Completed;

  20.                 SocketClientInfo clientInfo = new SocketClientInfo();
  21.                 socketEventArgs.UserToken = clientInfo;
  22.                 clientInfo.PeerIp = peerIp;
  23.                 clientInfo.PeerPort = peerPort;
  24.                 clientInfo.Tag = tag;

  26.                 bool willRaiseEvent = socket.ConnectAsync(socketEventArgs);
  27.                 if (!willRaiseEvent)
  28.                 {
  29.                     ProcessConnect(socketEventArgs);
  30.                     socketEventArgs.Completed -= SocketConnect_Completed;
  31.                     socketEventArgs.Dispose();
  32.                 }
  33.                 return true;
  34.             }
  35.             catch (Exception ex)
  36.             {
  37.                 NetLogger.Log("ConnectAsyn",ex);
  38.                 return false;
  39.             }
  40.         }

  42.         private void SocketConnect_Completed(object sender, SocketAsyncEventArgs socketEventArgs)
  43.         {
  44.             ProcessConnect(socketEventArgs);
  45.             socketEventArgs.Completed -= SocketConnect_Completed;
  46.             socketEventArgs.Dispose();
  47.         }

  49.         private void ProcessConnect(SocketAsyncEventArgs socketEventArgs)
  50.         {
  51.             SocketClientInfo clientInfo = socketEventArgs.UserToken as SocketClientInfo;
  52.             if (socketEventArgs.SocketError == SocketError.Success)
  53.             {
  54.                 DealConnectSocket(socketEventArgs.ConnectSocket, clientInfo);
  55.             }
  56.             else
  57.             {
  58.                 SocketEventParam socketParam = new SocketEventParam(EN_SocketEvent.connect, null);
  59.                 socketParam.ClientInfo = clientInfo;
  60.                 OnSocketConnectEvent?.Invoke(socketParam, null);
  61.             }
  62.         }


  65.         void DealConnectSocket(Socket socket, SocketClientInfo clientInfo)
  66.         {
  67.             clientInfo.SetClientInfo(socket);

  69.             AsyncSocketClient client = new AsyncSocketClient(socket);
  70.             client.SetClientInfo(clientInfo);

  72.             //触发事件
  73.             SocketEventParam socketParam = new SocketEventParam(EN_SocketEvent.connect, socket);
  74.             socketParam.ClientInfo = clientInfo;
  75.             OnSocketConnectEvent?.Invoke(socketParam, client);
  76.         }

  78.         public bool Connect(string peerIp, int peerPort, object tag, out Socket socket)
  79.         {
  80.             socket = null;
  81.             try
  82.             {
  83.                 Socket socketTmp = new Socket(SocketType.Stream, ProtocolType.Tcp);

  85.                 SocketClientInfo clientInfo = new SocketClientInfo();
  86.                 clientInfo.PeerIp = peerIp;
  87.                 clientInfo.PeerPort = peerPort;
  88.                 clientInfo.Tag = tag;

  90.                 EndPoint remoteEP = new IPEndPoint(IPAddress.Parse(peerIp), peerPort);
  91.                 socketTmp.Connect(remoteEP);
  92.                 if (!socketTmp.Connected)
  93.                     return false;

  95.                 DealConnectSocket(socketTmp, clientInfo);
  96.                 socket = socketTmp;
  97.                 return true;
  98.             }
  99.             catch (Exception ex)
  100.             {
  101.                 NetLogger.Log(string.Format("连接对方:({0}:{1})出错！", peerIp, peerPort), ex);
  102.                 return false;
  103.             }
  104.         }
  105.     }
  106. }

AsyncSocketClient socket收发处理

  1. using System;
  2. using System.Collections.Generic;
  3. using System.Diagnostics;
  4. using System.Net;
  5. using System.Net.Sockets;

  7. namespace IocpCore
  8. {
  9.     public class AsyncSocketClient
  10.     {
  11.         public static int IocpReadLen = 1024;

  13.         public readonly Socket ConnectSocket;

  15.         protected SocketAsyncEventArgs m_receiveEventArgs;
  16.         public SocketAsyncEventArgs ReceiveEventArgs { get { return m_receiveEventArgs; } set { m_receiveEventArgs = value; } }
  17.         protected byte[] m_asyncReceiveBuffer;

  19.         protected SocketAsyncEventArgs m_sendEventArgs;
  20.         public SocketAsyncEventArgs SendEventArgs { get { return m_sendEventArgs; } set { m_sendEventArgs = value; } }
  21.         protected byte[] m_asyncSendBuffer;

  23.         public event Action<AsyncSocketClient, byte[]> OnReadData;
  24.         public event Action<AsyncSocketClient, int> OnSendData;
  25.         public event Action<AsyncSocketClient> OnSocketClose;

  27.         static object releaseLock = new object();
  28.         public static int createCount = 0;
  29.         public static int releaseCount = 0;

  31.         ~AsyncSocketClient()
  32.         {
  33.             lock (releaseLock)
  34.             {
  35.                 releaseCount++;
  36.             }
  37.         }

  39.         public AsyncSocketClient(Socket socket)
  40.         {
  41.             lock (releaseLock)
  42.             {
  43.                 createCount++;
  44.             }

  46.             ConnectSocket = socket;

  48.             m_receiveEventArgs = new SocketAsyncEventArgs();
  49.             m_asyncReceiveBuffer = new byte[IocpReadLen];
  50.             m_receiveEventArgs.AcceptSocket = ConnectSocket;
  51.             m_receiveEventArgs.Completed += ReceiveEventArgs_Completed;

  53.             m_sendEventArgs = new SocketAsyncEventArgs();
  54.             m_asyncSendBuffer = new byte[IocpReadLen * 2];
  55.             m_sendEventArgs.AcceptSocket = ConnectSocket;
  56.             m_sendEventArgs.Completed += SendEventArgs_Completed;
  57.         }

  59.         SocketClientInfo _clientInfo;

  61.         public SocketClientInfo ClientInfo
  62.         {
  63.             get
  64.             {
  65.                 return _clientInfo;
  66.             }
  67.         }

  69.         internal void CreateClientInfo(NetListener netListener)
  70.         {
  71.             _clientInfo = new SocketClientInfo();
  72.             try
  73.             {
  74.                 _clientInfo.Tag = netListener._listenParam._tag;
  75.                 IPEndPoint ip = ConnectSocket.LocalEndPoint as IPEndPoint;
  76.                 Debug.Assert(netListener._listenParam._port == ip.Port);

  78.                 _clientInfo.LocalIp = ip.Address.ToString();
  79.                 _clientInfo.LocalPort = netListener._listenParam._port;

  81.                 ip = ConnectSocket.RemoteEndPoint as IPEndPoint;
  82.                 _clientInfo.PeerIp = ip.Address.ToString();
  83.                 _clientInfo.PeerPort = ip.Port;
  84.             }
  85.             catch (Exception ex)
  86.             {
  87.                 NetLogger.Log("CreateClientInfo", ex);
  88.             }
  89.         }
  90.         internal void SetClientInfo(SocketClientInfo clientInfo)
  91.         {
  92.             _clientInfo = clientInfo;
  93.         }

  95.         #region read process
  96.         bool _inReadPending = false;
  97.         public EN_SocketReadResult ReadNextData()
  98.         {
  99.             lock (this)
  100.             {
  101.                 if (_socketError)
  102.                     return EN_SocketReadResult.ReadError;
  103.                 if (_inReadPending)
  104.                     return EN_SocketReadResult.InAsyn;
  105.                 if(!ConnectSocket.Connected)
  106.                 {
  107.                     OnReadError();
  108.                     return EN_SocketReadResult.ReadError;
  109.                 }

  111.                 try
  112.                 {
  113.                     m_receiveEventArgs.SetBuffer(m_asyncReceiveBuffer, 0, m_asyncReceiveBuffer.Length);
  114.                     _inReadPending = true;
  115.                     bool willRaiseEvent = ConnectSocket.ReceiveAsync(ReceiveEventArgs); //投递接收请求
  116.                     if (!willRaiseEvent)
  117.                     {
  118.                         _inReadPending = false;
  119.                         ProcessReceive();
  120.                         if (_socketError)
  121.                         {
  122.                             OnReadError();
  123.                             return EN_SocketReadResult.ReadError;
  124.                         }
  125.                         return EN_SocketReadResult.HaveRead;
  126.                     }
  127.                     else
  128.                     {
  129.                         return EN_SocketReadResult.InAsyn;
  130.                     }
  131.                 }
  132.                 catch (Exception ex)
  133.                 {
  134.                     NetLogger.Log("ReadNextData", ex);
  135.                     _inReadPending = false;
  136.                     OnReadError();
  137.                     return EN_SocketReadResult.ReadError;
  138.                 }
  139.             }
  140.         }

  142.         private void ProcessReceive()
  143.         {
  144.             if (ReceiveEventArgs.BytesTransferred > 0
  145.                 && ReceiveEventArgs.SocketError == SocketError.Success)
  146.             {
  147.                 int offset = ReceiveEventArgs.Offset;
  148.                 int count = ReceiveEventArgs.BytesTransferred;

  150.                 byte[] readData = new byte[count];
  151.                 Array.Copy(m_asyncReceiveBuffer, offset, readData, 0, count);

  153.                 _inReadPending = false;
  154.                 if (!_socketError)
  155.                     OnReadData?.Invoke(this, readData);
  156.             }
  157.             else
  158.             {
  159.                 _inReadPending = false;
  160.                 OnReadError();
  161.             }
  162.         }

  164.         private void ReceiveEventArgs_Completed(object sender, SocketAsyncEventArgs e)
  165.         {
  166.             lock (this)
  167.             {
  168.                 _inReadPending = false;
  169.                 ProcessReceive();
  170.                 if (_socketError)
  171.                 {
  172.                     OnReadError();
  173.                 }
  174.             }
  175.         }

  177.         bool _socketError = false;
  178.         private void OnReadError()
  179.         {
  180.             lock (this)
  181.             {
  182.                 if (_socketError == false)
  183.                 {
  184.                     _socketError = true;
  185.                     OnSocketClose?.Invoke(this);
  186.                 }
  187.                 CloseClient();
  188.             }
  189.         }
  190.         #endregion

  192.         #region send process
  193.         int _sendBufferByteCount = 102400;
  194.         public int SendBufferByteCount
  195.         {
  196.             get
  197.             {
  198.                 return _sendBufferByteCount;
  199.             }
  200.             set
  201.             {
  202.                 if (value < 1024)
  203.                 {
  204.                     _sendBufferByteCount = 1024;
  205.                 }
  206.                 else
  207.                 {
  208.                     _sendBufferByteCount = value;
  209.                 }
  210.             }
  211.         }

  213.         SendBufferPool _sendDataPool = new SendBufferPool();
  214.         internal EN_SendDataResult PutSendData(byte[] data)
  215.         {
  216.            //此处省略302         }

  218.         private void SendEventArgs_Completed(object sender, SocketAsyncEventArgs sendEventArgs)
  219.         {
  220.             lock (this)
  221.             {
  222.                 try
  223.                 {
  224.                     _inSendPending = false;
  225.                     ProcessSend(m_sendEventArgs);

  227.                     int sendCount = 0;
  228.                     if (sendEventArgs.SocketError == SocketError.Success)
  229.                     {
  230.                         sendCount = sendEventArgs.BytesTransferred;
  231.                     }
  232.                     OnSendData?.Invoke(this, sendCount);

  234.                     if (_socketError)
  235.                     {
  236.                         OnSendError();
  237.                     }
  238.                 }
  239.                 catch (Exception ex)
  240.                 {
  241.                     NetLogger.Log("SendEventArgs_Completed", ex);
  242.                 }
  243.             }
  244.         }

  246.         private bool ProcessSend(SocketAsyncEventArgs sendEventArgs)
  247.         {
  248.             if (sendEventArgs.SocketError == SocketError.Success)
  249.             {
  250.                 return true;
  251.             }
  252.             else
  253.             {
  254.                 OnSendError();
  255.                 return false;
  256.             }
  257.         }

  259.         private int GetSendData()
  260.         {
  261.             int dataLen = 0;
  262.             while (true)
  263.             {
  264.                 byte[] data = _sendDataPool.GetObj();
  265.                 if (data == null)
  266.                     return dataLen;
  267.                 Array.Copy(data, 0, m_asyncSendBuffer, dataLen, data.Length);
  268.                 dataLen += data.Length;
  269.                 if (dataLen > IocpReadLen)
  270.                     break;
  271.             }
  272.             return dataLen;
  273.         }
  274.         private void OnSendError()
  275.         {
  276.             lock (this)
  277.             {
  278.                 if (_socketError == false)
  279.                 {
  280.                     _socketError = true;
  281.                     OnSocketClose?.Invoke(this);
  282.                 }
  283.                 CloseClient();
  284.             }
  285.         }
  286.         #endregion

  288.         internal void CloseSocket()
  289.         {
  290.             try
  291.             {
  292.                 ConnectSocket.Close();
  293.             }
  294.             catch (Exception ex)
  295.             {
  296.                 NetLogger.Log("CloseSocket", ex);
  297.             }
  298.         }

  300.         static object socketCloseLock = new object();
  301.         public static int closeSendCount = 0;
  302.         public static int closeReadCount = 0;

  304.         bool _disposeSend = false;
  305.         void CloseSend()
  306.         {
  307.             if (!_disposeSend && !_inSendPending)
  308.             {
  309.                 lock (socketCloseLock)
  310.                     closeSendCount++;

  312.                 _disposeSend = true;
  313.                 m_sendEventArgs.SetBuffer(null, 0, 0);
  314.                 m_sendEventArgs.Completed -= SendEventArgs_Completed;
  315.                 m_sendEventArgs.Dispose();
  316.             }
  317.         }

  319.         bool _disposeRead = false;
  320.         void CloseRead()
  321.         {
  322.             if (!_disposeRead && !_inReadPending)
  323.             {
  324.                 lock (socketCloseLock)
  325.                     closeReadCount++;

  327.                 _disposeRead = true;
  328.                 m_receiveEventArgs.SetBuffer(null, 0, 0);
  329.                 m_receiveEventArgs.Completed -= ReceiveEventArgs_Completed;
  330.                 m_receiveEventArgs.Dispose();
  331.             }
  332.         }
  333.         private void CloseClient()
  334.         {
  335.             try
  336.             {
  337.                 CloseSend();
  338.                 CloseRead();
  339.                 ConnectSocket.Close();
  340.             }
  341.             catch (Exception ex)
  342.             {
  343.                 NetLogger.Log("CloseClient", ex);
  344.             }
  345.         }

  347.         //发送缓冲大小
  348.         private List<byte[]> SplitData(byte[] data, int maxLen)
  349.         {
  350.             List<byte[]> items = new List<byte[]>();

  352.             int start = 0;
  353.             while (true)
  354.             {
  355.                 int itemLen = Math.Min(maxLen, data.Length - start);
  356.                 if (itemLen == 0)
  357.                     break;
  358.                 byte[] item = new byte[itemLen];
  359.                 Array.Copy(data, start, item, 0, itemLen);
  360.                 items.Add(item);

  362.                 start += itemLen;
  363.             }
  364.             return items;
  365.         }
  366.     }

  368.     public enum EN_SocketReadResult
  369.     {
  370.         InAsyn,
  371.         HaveRead,
  372.         ReadError
  373.     }

  375.     public enum EN_SocketSendResult
  376.     {
  377.         InAsyn,
  378.         HaveSend,
  379.         NoSendData,
  380.         SendError
  381.     }

  383.     class SendBufferPool
  384.     {
  385.         ObjectPool<byte[]> _bufferPool = new ObjectPool<byte[]>();

  387.         public Int64 _bufferByteCount = 0;
  388.         public bool PutObj(byte[] obj)
  389.         {
  390.             if (_bufferPool.PutObj(obj))
  391.             {
  392.                 lock (this)
  393.                 {
  394.                     _bufferByteCount += obj.Length;
  395.                 }
  396.                 return true;
  397.             }
  398.             else
  399.             {
  400.                 return false;
  401.             }
  402.         }

  404.         public byte[] GetObj()
  405.         {
  406.             byte[] result = _bufferPool.GetObj();
  407.             if (result != null)
  408.             {
  409.                 lock (this)
  410.                 {
  411.                     _bufferByteCount -= result.Length;
  412.                 }
  413.             }
  414.             return result;
  415.         }
  416.     }
  417. }

NetServer  聚合其他类

  1. using System;
  2. using System.Collections.Generic;
  3. using System.Diagnostics;
  4. using System.Linq;
  5. using System.Net.Sockets;
  6. using System.Threading;

  8. namespace IocpCore
  9. {
  10.     public class NetServer
  11.     {
  12.         public Action<SocketEventParam> OnSocketPacketEvent;

  14.         //每个连接发送缓冲大小
  15.         public int SendBufferBytePerClient { get; set; } = 1024 * 100;

  17.         bool _serverStart = false;
  18.         List<NetListener> _listListener = new List<NetListener>();

  20.         //负责对收到的字节流 组成完成的包
  21.         ClientPacketManage _clientPacketManage;

  23.         public Int64 SendByteCount { get; set; }
  24.         public Int64 ReadByteCount { get; set; }

  26.         List<ListenParam> _listListenPort = new List<ListenParam>();
  27.         public void AddListenPort(int port, object tag)
  28.         {
  29.             _listListenPort.Add(new ListenParam(port, tag));
  30.         }
  31.         /// <summary>
  32.         /// 
  33.         /// </summary>
  34.         /// <param name="listenFault">监听失败的端口</param>
  35.         /// <returns></returns>
  36.         public bool StartListen(out List<int> listenFault)
  37.         {
  38.             _serverStart = true;

  40.             _clientPacketManage = new ClientPacketManage(this);
  41.             _clientPacketManage.OnSocketPacketEvent += PutClientPacket;

  43.             _netConnectManage.OnSocketConnectEvent += SocketConnectEvent;

  45.             _listListener.Clear();
  46.             Thread thread1 = new Thread(new ThreadStart(NetPacketProcess));
  47.             thread1.Start();

  49.             Thread thread2 = new Thread(new ThreadStart(NetSendProcess));
  50.             thread2.Start();

  52.             Thread thread3 = new Thread(new ThreadStart(NetReadProcess));
  53.             thread3.Start();

  55.             listenFault = new List<int>();
  56.             foreach (ListenParam param in _listListenPort)
  57.             {
  58.                 NetListener listener = new NetListener(this);
  59.                 listener._listenParam = param;
  60.                 listener.OnAcceptSocket += Listener_OnAcceptSocket;
  61.                 if (!listener.StartListen())
  62.                 {
  63.                     listenFault.Add(param._port);
  64.                 }
  65.                 else
  66.                 {
  67.                     _listListener.Add(listener);
  68.                     NetLogger.Log(string.Format("监听成功！端口:{0}", param._port));
  69.                 }
  70.             }

  72.             return listenFault.Count == 0;
  73.         }

  75.         public void PutClientPacket(SocketEventParam param)
  76.         {
  77.             OnSocketPacketEvent?.Invoke(param);
  78.         }

  80.         //获取包的最小长度
  81.         int _packetMinLen;
  82.         int _packetMaxLen;
  83.         public int PacketMinLen
  84.         {
  85.             get { return _packetMinLen; }
  86.         }
  87.         public int PacketMaxLen
  88.         {
  89.             get { return _packetMaxLen; }
  90.         }

  92.         /// <summary>
  93.         /// 设置包的最小和最大长度
  94.         /// 当minLen=0时，认为是接收字节流
  95.         /// </summary>
  96.         /// <param name="minLen"></param>
  97.         /// <param name="maxLen"></param>
  98.         public void SetPacketParam(int minLen, int maxLen)
  99.         {
  100.             Debug.Assert(minLen >= 0);
  101.             Debug.Assert(maxLen > minLen);
  102.             _packetMinLen = minLen;
  103.             _packetMaxLen = maxLen;
  104.         }

  106.         //获取包的总长度
  107.         public delegate int delegate_GetPacketTotalLen(byte[] data, int offset);
  108.         public delegate_GetPacketTotalLen GetPacketTotalLen_Callback;

  110.         ObjectPoolWithEvent<SocketEventParam> _socketEventPool = new ObjectPoolWithEvent<SocketEventParam>();
  111.         private void NetPacketProcess()
  112.         {
  113.             while (_serverStart)
  114.             {
  115.                 try
  116.                 {
  117.                     DealEventPool();
  118.                 }
  119.                 catch (Exception ex)
  120.                 {
  121.                     NetLogger.Log(string.Format("DealEventPool 异常 {0}***{1}", ex.Message, ex.StackTrace));
  122.                 }
  123.                 _socketEventPool.WaitOne(1000);
  124.             }
  125.         }

  127.         Dictionary<Socket, AsyncSocketClient> _clientGroup = new Dictionary<Socket, AsyncSocketClient>();
  128.         public int ClientCount
  129.         {
  130.             get
  131.             {
  132.                 lock (_clientGroup)
  133.                 {
  134.                     return _clientGroup.Count;
  135.                 }
  136.             }
  137.         }
  138.         public List<Socket> ClientList
  139.         {
  140.             get
  141.             {
  142.                 lock (_clientGroup)
  143.                 {
  144.                     return _clientGroup.Keys.ToList();
  145.                 }
  146.             }
  147.         }

  149.         private void DealEventPool()
  150.         {
  151.             while (true)
  152.             {
  153.                 SocketEventParam param = _socketEventPool.GetObj();
  154.                 if (param == null)
  155.                     return;

  157.                 if (param.SocketEvent == EN_SocketEvent.close)
  158.                 {
  159.                     lock (_clientGroup)
  160.                     {
  161.                         _clientGroup.Remove(param.Socket);
  162.                     }
  163.                 }

  165.                 if (_packetMinLen == 0)//字节流处理
  166.                 {
  167.                     OnSocketPacketEvent?.Invoke(param);
  168.                 }
  169.                 else
  170.                 {
  171.                     //组成一个完整的包 逻辑
  172.                     _clientPacketManage.PutSocketParam(param);
  173.                 }
  174.             }
  175.         }

  177.         private void SocketConnectEvent(SocketEventParam param, AsyncSocketClient client)
  178.         {
  179.             try
  180.             {
  181.                 if (param.Socket == null || client == null) //连接失败
  182.                 {

  184.                 }
  185.                 else
  186.                 {
  187.                     lock (_clientGroup)
  188.                     {
  189.                         bool remove = _clientGroup.Remove(client.ConnectSocket);
  190.                         Debug.Assert(!remove);
  191.                         _clientGroup.Add(client.ConnectSocket, client);
  192.                     }

  194.                     client.OnSocketClose += Client_OnSocketClose;
  195.                     client.OnReadData += Client_OnReadData;
  196.                     client.OnSendData += Client_OnSendData;

  198.                     _listReadEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.read));
  199.                 }
  200.                 _socketEventPool.PutObj(param);
  201.             }
  202.             catch (Exception ex)
  203.             {
  204.                 NetLogger.Log(string.Format("SocketConnectEvent 异常 {0}***{1}", ex.Message, ex.StackTrace));
  205.             }
  206.         }

  208.         internal void OnRcvPacketLenError(Socket socket, byte[] buffer, int offset, int packetLen)
  209.         {
  210.             try
  211.             {
  212.                 lock (_clientGroup)
  213.                 {
  214.                     if (!_clientGroup.ContainsKey(socket))
  215.                     {
  216.                         Debug.Assert(false);
  217.                         return;
  218.                     }

  220.                     NetLogger.Log(string.Format("报长度异常!包长:{0}", packetLen));
  221.                     AsyncSocketClient client = _clientGroup[socket];
  222.                     client.CloseSocket();
  223.                 }
  224.             }
  225.             catch (Exception ex)
  226.             {
  227.                 NetLogger.Log(string.Format("OnRcvPacketLenError 异常 {0}***{1}", ex.Message, ex.StackTrace));
  228.             }
  229.         }

  231.         #region listen port
  232.         private void Listener_OnAcceptSocket(ListenParam listenPatam, AsyncSocketClient client)
  233.         {
  234.             try
  235.             {
  236.                 lock (_clientGroup)
  237.                 {
  238.                     bool remove = _clientGroup.Remove(client.ConnectSocket);
  239.                     Debug.Assert(!remove);
  240.                     _clientGroup.Add(client.ConnectSocket, client);
  241.                 }

  243.                 client.OnSocketClose += Client_OnSocketClose;
  244.                 client.OnReadData += Client_OnReadData;
  245.                 client.OnSendData += Client_OnSendData;

  247.                 _listReadEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.read));

  249.                 SocketEventParam param = new SocketEventParam(EN_SocketEvent.accept, client.ConnectSocket);
  250.                 param.ClientInfo = client.ClientInfo;

  252.                 _socketEventPool.PutObj(param);
  253.             }
  254.             catch (Exception ex)
  255.             {
  256.                 NetLogger.Log(string.Format("Listener_OnAcceptSocket 异常 {0}***{1}", ex.Message, ex.StackTrace));
  257.             }
  258.         }


  261.         ObjectPoolWithEvent<SocketEventDeal> _listSendEvent = new ObjectPoolWithEvent<SocketEventDeal>();
  262.         private void NetSendProcess()
  263.         {
  264.             while (true)
  265.             {
  266.                 DealSendEvent();
  267.                 _listSendEvent.WaitOne(1000);
  268.             }
  269.         }

  271.         ObjectPoolWithEvent<SocketEventDeal> _listReadEvent = new ObjectPoolWithEvent<SocketEventDeal>();
  272.         private void NetReadProcess()
  273.         {
  274.             while (true)
  275.             {
  276.                 DealReadEvent();
  277.                 _listReadEvent.WaitOne(1000);
  278.             }
  279.         }


  282.         private void DealSendEvent()
  283.         {
  284.             while (true)
  285.             {
  286.                 SocketEventDeal item = _listSendEvent.GetObj();
  287.                 if (item == null)
  288.                     break;
  289.                 switch (item.SocketEvent)
  290.                 {
  291.                     case EN_SocketDealEvent.send:
  292.                         {
  293.                             while (true)
  294.                             {
  295.                                 EN_SocketSendResult result = item.Client.SendNextData();
  296.                                 if (result == EN_SocketSendResult.HaveSend)
  297.                                     continue;
  298.                                 else
  299.                                     break;
  300.                             }
  301.                         }
  302.                         break;
  303.                     case EN_SocketDealEvent.read:
  304.                         {
  305.                             Debug.Assert(false);
  306.                         }
  307.                         break;                   
  308.                 }
  309.             }
  310.         }

  312.         private void DealReadEvent()
  313.         {
  314.             while (true)
  315.             {
  316.                 SocketEventDeal item = _listReadEvent.GetObj();
  317.                 if (item == null)
  318.                     break;
  319.                 switch (item.SocketEvent)
  320.                 {
  321.                     case EN_SocketDealEvent.read:
  322.                         {
  323.                             while (true)
  324.                             {
  325.                                 EN_SocketReadResult result = item.Client.ReadNextData();
  326.                                 if (result == EN_SocketReadResult.HaveRead)
  327.                                     continue;
  328.                                 else
  329.                                     break;
  330.                             }
  331.                         }
  332.                         break;
  333.                     case EN_SocketDealEvent.send:
  334.                         {
  335.                             Debug.Assert(false);
  336.                         }
  337.                         break;
  338.                 }
  339.             }
  340.         }

  342.         private void Client_OnReadData(AsyncSocketClient client, byte[] readData)
  343.         {
  344.             //读下一条
  345.             _listReadEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.read));

  347.             try
  348.             {
  349.                 SocketEventParam param = new SocketEventParam(EN_SocketEvent.read, client.ConnectSocket);
  350.                 param.ClientInfo = client.ClientInfo;
  351.                 param.Data = readData;
  352.                 _socketEventPool.PutObj(param);

  354.                 lock (this)
  355.                 {
  356.                     ReadByteCount += readData.Length;
  357.                 }
  358.             }
  359.             catch (Exception ex)
  360.             {
  361.                 NetLogger.Log(string.Format("Client_OnReadData 异常 {0}***{1}", ex.Message, ex.StackTrace));
  362.             }
  363.         }
  364. #endregion

  366.         private void Client_OnSendData(AsyncSocketClient client, int sendCount)
  367.         {
  368.             //发送下一条
  369.             _listSendEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.send));
  370.             lock (this)
  371.             {
  372.                 SendByteCount += sendCount;
  373.             }
  374.         }

  376.         private void Client_OnSocketClose(AsyncSocketClient client)
  377.         {
  378.             try
  379.             {
  380.                 SocketEventParam param = new SocketEventParam(EN_SocketEvent.close, client.ConnectSocket);
  381.                 param.ClientInfo = client.ClientInfo;
  382.                 _socketEventPool.PutObj(param);
  383.             }
  384.             catch (Exception ex)
  385.             {
  386.                 NetLogger.Log(string.Format("Client_OnSocketClose 异常 {0}***{1}", ex.Message, ex.StackTrace));
  387.             }
  388.         }

  390.         /// <summary>
  391.         /// 放到发送缓冲
  392.         /// </summary>
  393.         /// <param name="socket"></param>
  394.         /// <param name="data"></param>
  395.         /// <returns></returns>
  396.         public EN_SendDataResult SendData(Socket socket, byte[] data)
  397.         {
  398.             if (socket == null)
  399.                 return EN_SendDataResult.no_client;
  400.             lock (_clientGroup)
  401.             {
  402.                 if (!_clientGroup.ContainsKey(socket))
  403.                     return EN_SendDataResult.no_client;
  404.                 AsyncSocketClient client = _clientGroup[socket];
  405.                 EN_SendDataResult result = client.PutSendData(data);
  406.                 if (result == EN_SendDataResult.ok)
  407.                 {
  408.                     //发送下一条
  409.                     _listSendEvent.PutObj(new SocketEventDeal(client, EN_SocketDealEvent.send));                  
  410.                 }
  411.                 return result;
  412.             }
  413.         }

  415.         /// <summary>
  416.         /// 设置某个连接的发送缓冲大小
  417.         /// </summary>
  418.         /// <param name="socket"></param>
  419.         /// <param name="byteCount"></param>
  420.         /// <returns></returns>
  421.         public bool SetClientSendBuffer(Socket socket, int byteCount)
  422.         {
  423.             lock (_clientGroup)
  424.             {
  425.                 if (!_clientGroup.ContainsKey(socket))
  426.                     return false;
  427.                 AsyncSocketClient client = _clientGroup[socket];
  428.                 client.SendBufferByteCount = byteCount;
  429.                 return true;
  430.             }
  431.         }


  434.         #region connect process
  435.         NetConnectManage _netConnectManage = new NetConnectManage();
  436.         /// <summary>
  437.         /// 异步连接一个客户端
  438.         /// </summary>
  439.         /// <param name="peerIp"></param>
  440.         /// <param name="peerPort"></param>
  441.         /// <param name="tag"></param>
  442.         /// <returns></returns>
  443.         public bool ConnectAsyn(string peerIp, int peerPort, object tag)
  444.         {
  445.             return _netConnectManage.ConnectAsyn(peerIp, peerPort, tag);
  446.         }

  448.         /// <summary>
  449.         /// 同步连接一个客户端
  450.         /// </summary>
  451.         /// <param name="peerIp"></param>
  452.         /// <param name="peerPort"></param>
  453.         /// <param name="tag"></param>
  454.         /// <param name="socket"></param>
  455.         /// <returns></returns>
  456.         public bool Connect(string peerIp, int peerPort, object tag, out Socket socket)
  457.         {
  458.             return _netConnectManage.Connect(peerIp, peerPort, tag, out socket);
  459.         }
  460.         #endregion
  461.     }

  463.     enum EN_SocketDealEvent
  464.     {
  465.         read,
  466.         send,
  467.     }
  468.     class SocketEventDeal
  469.     {
  470.         public AsyncSocketClient Client { get; set; }
  471.         public EN_SocketDealEvent SocketEvent { get; set; }
  472.         public SocketEventDeal(AsyncSocketClient client, EN_SocketDealEvent socketEvent)
  473.         {
  474.             Client = client;
  475.             SocketEvent = socketEvent;
  476.         }
  477.     }
  478. }

库的使用

使用起来非常简单，示例如下 

 

  1. using IocpCore;
  2. using System;
  3. using System.Collections.Generic;
  4. using System.Linq;
  5. using System.Net.Sockets;
  6. using System.Text;
  7. using System.Threading.Tasks;
  8. using System.Windows;

  10. namespace WarningClient
  11. {
  12.     public class SocketServer
  13.     {
  14.         public Action<SocketEventParam> OnSocketEvent;

  16.         public Int64 SendByteCount
  17.         {
  18.             get
  19.             {
  20.                 if (_netServer == null)
  21.                     return 0;
  22.                 return _netServer.SendByteCount;
  23.             }
  24.         }
  25.         public Int64 ReadByteCount
  26.         {
  27.             get
  28.             {
  29.                 if (_netServer == null)
  30.                     return 0;
  31.                 return _netServer.ReadByteCount;
  32.             }
  33.         }

  35.         NetServer _netServer;
  36.         EN_PacketType _packetType = EN_PacketType.byteStream;
  37.         public void SetPacktType(EN_PacketType packetType)
  38.         {
  39.             _packetType = packetType;
  40.             if (_netServer == null)
  41.                 return;
  42.             if (packetType == EN_PacketType.byteStream)
  43.             {
  44.                 _netServer.SetPacketParam(0, 1024);
  45.             }
  46.             else
  47.             {
  48.                 _netServer.SetPacketParam(9, 1024);
  49.             }
  50.         }

  52.         public bool Init(List<int> listenPort)
  53.         {
  54.             NetLogger.OnLogEvent += NetLogger_OnLogEvent;
  55.             _netServer = new NetServer();
  56.             SetPacktType(_packetType);
  57.             _netServer.GetPacketTotalLen_Callback += GetPacketTotalLen;
  58.             _netServer.OnSocketPacketEvent += SocketPacketDeal;

  60.             foreach (int n in listenPort)
  61.             {
  62.                 _netServer.AddListenPort(n, n);
  63.             }

  65.             List<int> listenFault;
  66.             bool start = _netServer.StartListen(out listenFault);
  67.             return start;
  68.         }

  70.         int GetPacketTotalLen(byte[] data, int offset)
  71.         {
  72.             if (MainWindow._packetType == EN_PacketType.znss)
  73.                 return GetPacketZnss(data, offset);
  74.             else
  75.                 return GetPacketAnzhiyuan(data, offset);
  76.         }

  78.         int GetPacketAnzhiyuan(byte[] data, int offset)
  79.         {
  80.             int n = data[offset + 5] + 6;
  81.             return n;
  82.         }

  84.         int GetPacketZnss(byte[] data, int offset)
  85.         {
  86.             int packetLen = (int)(data[4]) + 5;
  87.             return packetLen;
  88.         }


  91.         public bool ConnectAsyn(string peerIp, int peerPort, object tag)
  92.         {
  93.             return _netServer.ConnectAsyn(peerIp, peerPort, tag);
  94.         }

  96.         public bool Connect(string peerIp, int peerPort, object tag, out Socket socket)
  97.         {
  98.             return _netServer.Connect(peerIp, peerPort, tag, out socket);
  99.         }

  101.         private void NetLogger_OnLogEvent(string message)
  102.         {
  103.             AppLog.Log(message);
  104.         }

  106.         Dictionary<Socket, SocketEventParam> _clientGroup = new Dictionary<Socket, SocketEventParam>();

  108.         public int ClientCount
  109.         {
  110.             get
  111.             {
  112.                 lock (_clientGroup)
  113.                 {
  114.                     return _clientGroup.Count;
  115.                 }
  116.             }
  117.         }
  118.         public List<Socket> ClientList
  119.         {
  120.             get
  121.             {
  122.                 if (_netServer != null)
  123.                     return _netServer.ClientList;
  124.                 return new List<Socket>();
  125.             }
  126.         }
  127.         void AddClient(SocketEventParam socketParam)
  128.         {
  129.             lock (_clientGroup)
  130.             {
  131.                 _clientGroup.Remove(socketParam.Socket);
  132.                 _clientGroup.Add(socketParam.Socket, socketParam);
  133.             }
  134.         }

  136.         void RemoveClient(SocketEventParam socketParam)
  137.         {
  138.             lock (_clientGroup)
  139.             {
  140.                 _clientGroup.Remove(socketParam.Socket);
  141.             }
  142.         }

  144.         ObjectPool<SocketEventParam> _readDataPool = new ObjectPool<SocketEventParam>();

  146.         public ObjectPool<SocketEventParam> ReadDataPool
  147.         {
  148.             get
  149.             {
  150.                 return _readDataPool;
  151.             }
  152.         }

  154.         private void SocketPacketDeal(SocketEventParam socketParam)
  155.         {
  156.             OnSocketEvent?.Invoke(socketParam);
  157.             if (socketParam.SocketEvent == EN_SocketEvent.read)
  158.             {
  159.                 if (MainWindow._isShowReadPacket)
  160.                     _readDataPool.PutObj(socketParam);
  161.             }
  162.             else if (socketParam.SocketEvent == EN_SocketEvent.accept)
  163.             {
  164.                 AddClient(socketParam);
  165.                 string peerIp = socketParam.ClientInfo.PeerIpPort;
  166.                 AppLog.Log(string.Format("客户端链接!本地端口:{0},对端:{1}",
  167.                     socketParam.ClientInfo.LocalPort, peerIp));
  168.             }
  169.             else if (socketParam.SocketEvent == EN_SocketEvent.connect)
  170.             {
  171.                 string peerIp = socketParam.ClientInfo.PeerIpPort;
  172.                 if (socketParam.Socket != null)
  173.                 {
  174.                     AddClient(socketParam);

  176.                     AppLog.Log(string.Format("连接对端成功!本地端口:{0},对端:{1}",
  177.                        socketParam.ClientInfo.LocalPort, peerIp));
  178.                 }
  179.                 else
  180.                 {
  181.                     AppLog.Log(string.Format("连接对端失败!本地端口:{0},对端:{1}",
  182.                         socketParam.ClientInfo.LocalPort, peerIp));
  183.                 }
  184.             }
  185.             else if (socketParam.SocketEvent == EN_SocketEvent.close)
  186.             {
  187.                 MainWindow.MainWnd.OnSocketDisconnect(socketParam.Socket);
  188.                 RemoveClient(socketParam);
  189.                 string peerIp = socketParam.ClientInfo.PeerIpPort;
  190.                 AppLog.Log(string.Format("客户端断开!本地端口:{0},对端:{1},",
  191.                     socketParam.ClientInfo.LocalPort, peerIp));
  192.             }
  193.         }

  195.         public EN_SendDataResult SendData(Socket socket, byte[] data)
  196.         {
  197.             if(socket == null)
  198.             {
  199.                 MessageBox.Show("还没连接！");
  200.                 return EN_SendDataResult.no_client;
  201.             }
  202.             return _netServer.SendData(socket, data);
  203.         }

  205.         internal void SendToAll(byte[] data)
  206.         {
  207.             lock (_clientGroup)
  208.             {
  209.                 foreach (Socket socket in _clientGroup.Keys)
  210.                 {
  211.                     SendData(socket, data);
  212.                 }
  213.             }
  214.         }
  215.     }
  216. }

博客原文:https://www.cnblogs.com/yuanchenhui/p/asyn_scoket.html

 

    本站是提供个人知识管理的网络存储空间，所有内容均由用户发布，不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息，谨防诈骗。如发现有害或侵权内容，请点击一键举报。
    转藏 分享 献花（0）

    来自： 眼镜蛇图书馆zh > 《C#/Netcore》

    举报/认领

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多