高性能 socket 框架

      最近用C# 寫了一個易用，高性能的socket，抗住了ServerTestTool.exe 壓力測試工具 

下載地址  GitHub       QQ群: 664740973

https://gitee.com/fengma_312/socket.core

https://www.nuget.org/packages/socket.core/

socket.core

這是一個基於C# .net standard2.0 寫的socket框架，可使用於.net Framework/dotnet core程序集，能在window(IOCP)/linux(epoll)運行.使用異步連接，異步發送，異步接收，性能爆表，並且通過壓力測試。

安裝NuGet:
Package Manager: Install-Package socket.core
.Net CLI :dotnet add package socket.core
Paket CLI:paket add socket.core

服務端所在socket.core.Server命名空間下，分別為三種模式 push/pull/pack
客戶端所在socket.core.Client命名空間下，分別為三種模式 push/pull/pack

主要流程與對應的方法和事件介紹.
注:connectId(int)代表着一個連接對象,data(byte[]),success(bool)

• 1.初始化socket(對應的三種模式)

  實例化服務端類 TcpPushServer/TcpPullServer/TcpPackServer
  實例化客戶端類 TcpPushClient/TcpPullClient/TcpPackClient
  參數介紹int numConnections同時處理的最大連接數,int receiveBufferSize用於每個套接字I/O操作的緩沖區大小(接收端), int overtime超時時長,單位秒.(每10秒檢查一次)，當值為0時，不設置超時,uint headerFlag包頭標記范圍0~1023(0x3FF),當包頭標識等於0時，不校驗包頭

• 2.啟動監聽/連接服務器

  服務端 server.Start(port);
  客戶端 client.Connect(ip,port);

• 3.觸發連接事件

  服務端 server.OnAccept(connectId); 接收到一個連接id,可用他來發送，接收，關閉的標記
  客戶端 client.OnConnect(success); 接收是否成功連接到服務器

• 4.發送消息

  服務端 server.Send(connectId,data,offset,length);
  客戶端 client.Send(data,offset,length);

• 5.觸發已發送事件

  服務端 server.OnSend(connectId,length);
  客戶端 client.OnSend(length);

• 6.觸發接收事件

  服務端 server.OnReceive(connectId, data);
  客戶端 client.OnReceive(data);

• 7.關閉連接

  服務端 server.Close(connectId);
  客戶端 client.Close();

• 8.觸發關閉連接事件

  服務端 server.OnClose(connectId);
  客戶端 client.OnClose();

三種模型簡介

• 一:push

  當接收到數據時會觸發監聽事件OnReceive(connectId,data);把數據立馬“推”給應用程序

• 二:pull

  當接收到數據時會觸發監聽事件OnReceive(connectId,length)，告訴應用程序當前已經接收到了多少數據長度，應用程序可使用GetLength(connectId)方法檢查已接收的數據的長度，如果滿足則調用組件的Fetch(connectId,length)方法，把需要的數據“拉”出來

• 三:pack

  pack模型組件是push和pull模型的結合體，應用程序不必要處理分包/合包，組件保證每個server.OnReceive(connectId,data)/client.OnReceive(data)事件都向應用程序提供一個完整的數據包
  注：pack模型組件會對應用程序發送的每個數據包自動加上4個字節(32bit)的包頭，組件接收到數據時，根據包頭信息自動分包，每個完整的數據包通過OnReceive(connectId, data)事件發送給應用程序
  PACK包頭格式(4字節)4*8=32
  XXXXXXXXXXYYYYYYYYYYYYYYYYYYYYYY
  前10位X為包頭標識位，用於數據包校驗，有效包頭標識取值范圍0~1023(0x3FF),當包頭標識等於0時，不校驗包頭，后22位Y為長度位，記錄包體長度。有效數據包最大長度不能超過4194303（0x3FFFFF）字節(byte),應用程序可以通過TcpPackServer/TcpPackClient構造函數參數headerFlag設置

服務端其它方法介紹

• 1. bool SetAttached(int connectId, object data)

  服務端為每個客戶端設置附加數據，避免用戶自己再建立用戶映射表

• 2. T GetAttached(int connectId)

  獲取指定客戶端的附加數據

  2017/12/27

  技術在於分享,大家共同進步

 核心類 

using System;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Linq;
using System.Collections.Concurrent;
using socket.core.Common;

namespace socket.core.Server
{
    /// <summary>
    /// tcp Socket監聽基庫
    /// </summary>
    internal class TcpServer
    {
        /// <summary>
        /// 連接標示 自增長
        /// </summary>
        private int connectId;
        /// <summary>
        /// 同時處理的最大連接數
        /// </summary>
        private int m_numConnections;
        /// <summary>
        /// 用於每個套接字I/O操作的緩沖區大小
        /// </summary>
        private int m_receiveBufferSize;
        /// <summary>
        /// 所有套接字接收操作的一個可重用的大型緩沖區集合。
        /// </summary>
        private BufferManager m_bufferManager;
        /// <summary>
        /// 用於監聽傳入連接請求的套接字
        /// </summary>
        private Socket listenSocket;
        /// <summary>
        /// 接受端SocketAsyncEventArgs對象重用池，接受套接字操作
        /// </summary>
        private SocketAsyncEventArgsPool m_receivePool;
        /// <summary>
        /// 發送端SocketAsyncEventArgs對象重用池，發送套接字操作
        /// </summary>
        private SocketAsyncEventArgsPool m_sendPool;
        /// <summary>
        /// 超時，如果超時，服務端斷開連接，客戶端需要重連操作
        /// </summary>
        private int overtime;
        /// <summary>
        /// 超時檢查間隔時間(秒)
        /// </summary>
        private int overtimecheck = 1;
        /// <summary>
        /// 能接到最多客戶端個數的原子操作
        /// </summary>
        private Semaphore m_maxNumberAcceptedClients;
        /// <summary>
        /// 已經連接的對象池
        /// </summary>
        internal ConcurrentDictionary<int, ConnectClient> connectClient;
        /// <summary>
        /// 發送線程數
        /// </summary>
        private int sendthread = 10;
        /// <summary>
        /// 需要發送的數據
        /// </summary>
        private ConcurrentQueue<SendingQueue>[] sendQueues;      
        /// <summary>
        /// 鎖
        /// </summary>
        private Mutex mutex = new Mutex();
        /// <summary>
        /// 連接成功事件
        /// </summary>
        internal event Action<int> OnAccept;
        /// <summary>
        /// 接收通知事件
        /// </summary>
        internal event Action<int, byte[], int, int> OnReceive;
        /// <summary>
        /// 已送通知事件
        /// </summary>
        internal event Action<int, int> OnSend;
        /// <summary>
        /// 斷開連接通知事件
        /// </summary>
        internal event Action<int> OnClose;

        /// <summary>
        /// 設置基本配置
        /// </summary>   
        /// <param name="numConnections">同時處理的最大連接數</param>
        /// <param name="receiveBufferSize">用於每個套接字I/O操作的緩沖區大小(接收端)</param>
        /// <param name="overTime">超時時長,單位秒.(每10秒檢查一次)，當值為0時，不設置超時</param>
        internal TcpServer(int numConnections, int receiveBufferSize, int overTime)
        {
            overtime = overTime;
            m_numConnections = numConnections;
            m_receiveBufferSize = receiveBufferSize;
            m_bufferManager = new BufferManager(receiveBufferSize * m_numConnections, receiveBufferSize);
            m_receivePool = new SocketAsyncEventArgsPool(m_numConnections);
            m_sendPool = new SocketAsyncEventArgsPool(m_numConnections);
            m_maxNumberAcceptedClients = new Semaphore(m_numConnections, m_numConnections);
            Init();
        }

        /// <summary>
        /// 初始化服務器通過預先分配的可重復使用的緩沖區和上下文對象。這些對象不需要預先分配或重用，但這樣做是為了說明API如何可以易於用於創建可重用對象以提高服務器性能。
        /// </summary>
        private void Init()
        {
            connectClient = new ConcurrentDictionary<int, ConnectClient>();
            sendQueues = new ConcurrentQueue<SendingQueue>[sendthread];
            for (int i = 0; i < sendthread; i++)
            {
                sendQueues[i] = new ConcurrentQueue<SendingQueue>();
            }
            //分配一個大字節緩沖區，所有I/O操作都使用一個。這個侍衛對內存碎片
            m_bufferManager.InitBuffer();
            //預分配的接受對象池socketasynceventargs，並分配緩存
            SocketAsyncEventArgs saea_receive;
            //分配的發送對象池socketasynceventargs，但是不分配緩存
            SocketAsyncEventArgs saea_send;
            for (int i = 0; i < m_numConnections; i++)
            {
                //預先接受端分配一組可重用的消息
                saea_receive = new SocketAsyncEventArgs();
                saea_receive.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
                //分配緩沖池中的字節緩沖區的socketasynceventarg對象
                m_bufferManager.SetBuffer(saea_receive);
                m_receivePool.Push(saea_receive);
                //預先發送端分配一組可重用的消息
                saea_send = new SocketAsyncEventArgs();
                saea_send.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
                m_sendPool.Push(saea_send);
            }
        }

        /// <summary>
        /// 啟動tcp服務偵聽
        /// </summary>       
        /// <param name="port">監聽端口</param>
        internal void Start(int port)
        {
            IPEndPoint localEndPoint = new IPEndPoint(IPAddress.Any, port);
            //創建listens是傳入的套接字。
            listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            listenSocket.NoDelay = true;
            //綁定端口
            listenSocket.Bind(localEndPoint);
            //掛起的連接隊列的最大長度。
            listenSocket.Listen(1000);
            //在監聽套接字上接受
            StartAccept(null);
            //發送線程
            for (int i = 0; i < sendthread; i++)
            {
                Thread thread = new Thread(StartSend);
                thread.IsBackground = true;
                thread.Priority = ThreadPriority.AboveNormal;
                thread.Start(i);
            }
            //超時機制
            if (overtime > 0)
            {
                Thread heartbeat = new Thread(new ThreadStart(() =>
                {
                    Heartbeat();
                }));
                heartbeat.IsBackground = true;
                heartbeat.Priority = ThreadPriority.Lowest;
                heartbeat.Start();
            }
        }

        /// <summary>
        /// 超時機制
        /// </summary>
        private void Heartbeat()
        {
            //計算超時次數 ，超過count就當客戶端斷開連接。服務端清除該連接資源
            int count = overtime / overtimecheck;
            while (true)
            {
                foreach (var item in connectClient.Values)
                {
                    if (item.keep_alive >= count)
                    {
                        item.keep_alive = 0;
                        CloseClientSocket(item.saea_receive);
                    }
                }
                foreach (var item in connectClient.Values)
                {
                    item.keep_alive++;
                }
                Thread.Sleep(overtimecheck * 1000);
            }
        }

        #region Accept

        /// <summary>
        /// 開始接受客戶端的連接請求的操作。
        /// </summary>
        /// <param name="acceptEventArg">發布時要使用的上下文對象服務器偵聽套接字上的接受操作</param>
        private void StartAccept(SocketAsyncEventArgs acceptEventArg)
        {
            if (acceptEventArg == null)
            {
                acceptEventArg = new SocketAsyncEventArgs();
                acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
            }
            else
            {
                // 套接字必須被清除，因為上下文對象正在被重用。
                acceptEventArg.AcceptSocket = null;
            }
            m_maxNumberAcceptedClients.WaitOne();
            //准備一個客戶端接入
            if (!listenSocket.AcceptAsync(acceptEventArg))
            {
                ProcessAccept(acceptEventArg);
            }
        }

        /// <summary>
        /// 當異步連接完成時調用此方法
        /// </summary>
        /// <param name="e">操作對象</param>
        private void ProcessAccept(SocketAsyncEventArgs e)
        {
            connectId++;
            //把連接到的客戶端信息添加到集合中
            ConnectClient connecttoken = new ConnectClient();
            connecttoken.socket = e.AcceptSocket;
            //從接受端重用池獲取一個新的SocketAsyncEventArgs對象
            connecttoken.saea_receive = m_receivePool.Pop();
            connecttoken.saea_receive.UserToken = connectId;
            connecttoken.saea_receive.AcceptSocket = e.AcceptSocket;
            connectClient.TryAdd(connectId, connecttoken);
            //一旦客戶機連接，就准備接收。
            if (!e.AcceptSocket.ReceiveAsync(connecttoken.saea_receive))
            {
                ProcessReceive(connecttoken.saea_receive);
            }
            //事件回調
            if (OnAccept != null)
            {
                OnAccept(connectId);
            }
            //接受第二連接的請求
            StartAccept(e);
        }

        #endregion

        #region 接受處理 receive

        /// <summary>
        /// 接受處理回調
        /// </summary>
        /// <param name="e">操作對象</param>
        private void ProcessReceive(SocketAsyncEventArgs e)
        {
            //檢查遠程主機是否關閉連接
            if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
            {
                int connectId = (int)e.UserToken;
                ConnectClient client;
                if (!connectClient.TryGetValue(connectId, out client))
                {
                    return;
                }
                //如果接收到數據，超時記錄設置為0
                if (overtime > 0)
                {
                    if (client != null)
                    {
                        client.keep_alive = 0;
                    }
                }
                //回調               
                if (OnReceive != null)
                {
                    if (client != null)
                    {
                        OnReceive(connectId, e.Buffer, e.Offset, e.BytesTransferred);
                    }
                }
                //准備下次接收數據      
                try
                {
                    if (!e.AcceptSocket.ReceiveAsync(e))
                    {
                        ProcessReceive(e);
                    }
                }
                catch (ObjectDisposedException ex)
                {
                    if (OnClose != null)
                    {
                        OnClose(connectId);
                    }
                }
            }
            else
            {
                CloseClientSocket(e);
            }
        }

        #endregion

        #region 發送處理 send

        /// <summary>
        /// 開始啟用發送
        /// </summary>
        private void StartSend(object thread)
        {
            while (true)
            {
                SendingQueue sending;
                if (sendQueues[(int)thread].TryDequeue(out sending))
                {
                    Send(sending);
                }
                else
                {
                    Thread.Sleep(100);
                }
            }
        }

        /// <summary>
        /// 異步發送消息 
        /// </summary>
        /// <param name="connectId">連接ID</param>
        /// <param name="data">數據</param>
        /// <param name="offset">偏移位</param>
        /// <param name="length">長度</param>
        internal void Send(int connectId, byte[] data, int offset, int length)
        {
            sendQueues[connectId % sendthread].Enqueue(new SendingQueue() { connectId = connectId, data = data, offset = offset, length = length });
        }

        /// <summary>
        /// 異步發送消息 
        /// </summary>
        /// <param name="sendQuere">發送消息體</param>
        private void Send(SendingQueue sendQuere)
        {
            ConnectClient client;
            if (!connectClient.TryGetValue(sendQuere.connectId, out client))
            {
                return;
            }
            //如果發送池為空時，臨時新建一個放入池中
            mutex.WaitOne();
            if (m_sendPool.Count == 0)
            {
                SocketAsyncEventArgs saea_send = new SocketAsyncEventArgs();
                saea_send.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
                m_sendPool.Push(saea_send);
            }
            SocketAsyncEventArgs sendEventArgs = m_sendPool.Pop();
            mutex.ReleaseMutex();
            sendEventArgs.UserToken = sendQuere.connectId;
            sendEventArgs.SetBuffer(sendQuere.data, sendQuere.offset, sendQuere.length);
            try
            {
                if (!client.socket.SendAsync(sendEventArgs))
                {
                    ProcessSend(sendEventArgs);
                }
            }
            catch (ObjectDisposedException ex)
            {
                if (OnClose != null)
                {
                    OnClose(sendQuere.connectId);
                }
            }
            sendQuere = null;
        }

        /// <summary>
        /// 發送回調
        /// </summary>
        /// <param name="e">操作對象</param>
        private void ProcessSend(SocketAsyncEventArgs e)
        {
            if (e.SocketError == SocketError.Success)
            {
                m_sendPool.Push(e);
                if (OnSend != null)
                {
                    OnSend((int)e.UserToken, e.BytesTransferred);
                }
            }
            else
            {
                CloseClientSocket(e);
            }
        }

        #endregion

        /// <summary>
        /// 每當套接字上完成接收或發送操作時，都會調用此方法。
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e">與完成的接收操作關聯的SocketAsyncEventArg</param>
        private void IO_Completed(object sender, SocketAsyncEventArgs e)
        {
            //確定剛剛完成哪種類型的操作並調用相關的處理程序
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Receive:
                    ProcessReceive(e);
                    break;
                case SocketAsyncOperation.Send:
                    ProcessSend(e);
                    break;
                case SocketAsyncOperation.Accept:
                    ProcessAccept(e);
                    break;
                default:
                    break;
            }
        }

        #region 斷開連接處理


        /// <summary>
        /// 客戶端斷開一個連接
        /// </summary>
        /// <param name="connectId">連接標記</param>
        internal void Close(int connectId)
        {
            ConnectClient client;
            if (!connectClient.TryGetValue(connectId, out client))
            {
                return;
            }
            CloseClientSocket(client.saea_receive);
        }

        /// <summary>
        /// 斷開一個連接
        /// </summary>
        /// <param name="e">操作對象</param>
        private void CloseClientSocket(SocketAsyncEventArgs e)
        {
            if (e.LastOperation == SocketAsyncOperation.Receive)
            {
                int connectId = (int)e.UserToken;
                ConnectClient client;
                if (!connectClient.TryGetValue(connectId, out client))
                {
                    return;
                }
                if (client.socket.Connected == false)
                {
                    return;
                }
                try
                {
                    client.socket.Shutdown(SocketShutdown.Both);
                }
                // 拋出客戶端進程已經關閉
                catch (Exception) { }
                client.socket.Close();
                m_receivePool.Push(e);
                m_maxNumberAcceptedClients.Release();
                if (OnClose != null)
                {
                    OnClose(connectId);
                }
                connectClient.TryRemove((int)e.UserToken, out client);
                client = null;
            }
        }

        #endregion

        #region 附加數據

        /// <summary>
        /// 給連接對象設置附加數據
        /// </summary>
        /// <param name="connectId">連接標識</param>
        /// <param name="data">附加數據</param>
        /// <returns>true:設置成功,false:設置失敗</returns>
        internal bool SetAttached(int connectId, object data)
        {
            ConnectClient client;
            if (!connectClient.TryGetValue(connectId, out client))
            {
                return false;
            }
            client.attached = data;
            return true;
        }

        /// <summary>
        /// 獲取連接對象的附加數據
        /// </summary>
        /// <param name="connectId">連接標識</param>
        /// <returns>附加數據，如果沒有找到則返回null</returns>
        internal T GetAttached<T>(int connectId)
        {
            ConnectClient client;
            if (!connectClient.TryGetValue(connectId, out client))
            {
                return default(T);
            }
            else
            {
                return (T)client.attached;
            }
        }
        #endregion
    }

}