用C#實現TCP連接池

背景

   最近對接一個TCP協議，這個協議定義的非常好，有頭標識和校驗位！但是，接口提供方定的通信協議還是欠缺考慮...正常情況下，這個協議是沒有問題的，但是在高並發的情況下，客戶端方就需要點真功夫了。

分析

    該通信協議中，沒有使用事務號，也就是說，用同一條連接連續發送兩次請求，你不知道返回的響應數據是哪個請求的。你可能會說，第一個響應是第一個請求的，第二個響應是第二個請求的！這是絕對的理想情況，服務器處理所有請求的耗時一樣，網絡沒有抖動。如果耗時和網絡抖動都無法確定的情況下，響應順序與請求順序就有可能不一致！結論就是，如果只有一個連接，那么所有請求只能排隊，一個請求處理完才能發送下一個。那如果高並發請求比較多怎么辦？

    如何用沒有事務號的通信協議實現高並發？Http？bingo！Http的請求方式就是解決這個問題的榜樣！Http就是在一個連接中一次只發送一個請求，請求沒有收到響應或者超時斷開的情況下，是不會發送第二個請求的。那么Http是怎么並發的？通過同時發送多個Http請求來實現並發！以Http為榜樣，那么就可以通過一個TCP連接一次發送一個請求，這個請求沒有結束之前不再發送新的請求，如此來保證請求與響應的匹配，通過多個連接同時發送多個請求來實現高並發。

    解決思路有了，還得考慮下性能。肯定不能每次發送請求就新建一個連接，請求結束就斷開連接，這樣很不TCP！那就考慮復用TCP連接，請求未結束之前，這個連接不可用，請求結束后不斷開連接，可供新請求使用。並發量肯定有低峰和高峰，低峰的時候，不需要保留太多的連接；高峰的時候，如果不加以控制，TCP連接數量會飆升，也需要加以控制！

實現

    基於以上分析，需要寫一個TCP連接池，該連接池可配置最少、最多連接數量，以及連接可空閑時間。當需要發送請求時，如果沒有可用連接，並且池內連接數量不超過最大數量，就創建新的連接；當池內數量達到上限，並且需要發送請求時，需要阻塞，直至有可用的連接；當連接空閑時間達到設置的可空閑時間，並且池內連接數量大於最小值時，清理掉多余的連接。

    以下是連接池類，ConnectionInfo類和接口IConnection沒有發，但是不影響看連接池代碼。歡迎拍磚！

    /// <summary>
    /// 連接池
    /// 對於不能單連並發的，用此連接池實現多連接並發，可控制最少最大連接數量
    /// 連接保活
    /// </summary>
    public class ConnectionPool : IDisposable
    {
        private ConcurrentQueue<ConnectionInfo> m_IdleConnections;
        private ConcurrentDictionary<long, ConnectionInfo> m_Connections;
        private object m_ConnectionLock = new object();

        private int m_Max;
        private int m_Min;
        private TimeSpan m_Expired;
        private Func<IConnection> m_ConnectionFactory;

        private bool m_IsConnected = false;
        /// <summary>
        /// 是否連接
        /// </summary>
        public bool IsConnected
        {
            get
            {
                return m_IsConnected;
            }
            set
            {
                if (m_IsConnected != value)
                {
                    m_IsConnected = value;
                    if (m_IsConnected)
                    {
                        Connected?.Invoke(this, EventArgs.Empty);
                    }
                    else
                    {
                        Disconnected?.Invoke(this, EventArgs.Empty);
                    }
                }
            }
        }

        /// <summary>
        /// 構造
        /// </summary>
        /// <param name="minConnection">最小連接數</param>
        /// <param name="maxConnection">最大連接數</param>
        /// <param name="expired">過期時間 空閑時間超過此值則會被清理掉</param>
        /// <param name="connectionFactory">創建連接的回調</param>
        public ConnectionPool(int minConnection, int maxConnection, TimeSpan expired, Func<IConnection> connectionFactory)
        {
            m_Min = minConnection;
            m_Max = maxConnection;
            m_Expired = expired;
            m_ConnectionFactory = connectionFactory;
            m_Connections = new ConcurrentDictionary<long, ConnectionInfo>();
            m_IdleConnections = new ConcurrentQueue<ConnectionInfo>();
            StartClear();
        }

        private long m_No = 0;
        private object m_NoLock = new object();

        /// <summary>
        /// 生成新的編號
        /// </summary>
        /// <returns></returns>
        private long NewNo()
        {
            long no;
            lock (m_NoLock)
            {
                no = ++m_No;
            }
            return no;
        }

        /// <summary>
        /// 抓取連接
        /// </summary>
        /// <returns></returns>
        private ConnectionInfo GrabConnection()
        {
            //判斷空閑隊列是否有
            ConnectionInfo connectionInfo = null;
        //開始抓取連接
        Begin:
            while (!m_IsDisposed && !m_IdleConnections.IsEmpty)
            {
                if (m_IdleConnections.TryDequeue(out connectionInfo))
                {
                    if (m_Connections.ContainsKey(connectionInfo.No))
                    {//取到的連接沒有被銷毀
                        break;
                    }
                    else
                    {//不可用則銷毀此連接，繼續尋找
                        DestoryConnection(connectionInfo);
                        connectionInfo = null;
                    }
                }
                else
                {
                    connectionInfo = null;
                }
            }
            if (!m_IsDisposed && connectionInfo == null)
            {//沒有取到連接
                if (!CreateOrAddConnection(null))
                {//創建連接失敗，睡眠10ms
                    Thread.Sleep(10);
                }
                //繼續抓取連接
                goto Begin;
            }
            return connectionInfo;
        }

        /// <summary>
        /// 創建或者添加連接
        /// </summary>
        public bool CreateOrAddConnection(IConnection connection)
        {
            bool rst = false;
            if (m_Connections.Count < m_Max)
            {
                lock (m_ConnectionLock)
                {
                    if (m_Connections.Count < m_Max)
                    {
                        if (connection == null)
                        {
                            connection = m_ConnectionFactory.Invoke();
                        }
                        var conInfo = new ConnectionInfo()
                        {
                            No = NewNo(),
                            Connection = connection,
                            CreateTime = DateTime.Now
                        };
                        if (m_Connections.TryAdd(conInfo.No, conInfo))
                        {
                            connection.Connect();
                            IsConnected = connection.IsConnected;
                            m_IdleConnections.Enqueue(conInfo);
                            rst = true;
                            OutputDebugInfo(string.Format("創建{0}", conInfo.No));
                        }
                    }
                }
            }
            return rst;
        }

        /// <summary>
        /// 銷毀連接
        /// </summary>
        /// <param name="connectionInfo"></param>
        private void DestoryConnection(ConnectionInfo connectionInfo)
        {
            ConnectionInfo temp;
            while (m_Connections.ContainsKey(connectionInfo.No))
            {
                if (m_Connections.TryRemove(connectionInfo.No, out temp))
                {
                    break;
                }
                else
                {
                    Thread.Sleep(10);
                }
            }
            try
            {
                connectionInfo.Connection.Disconnect();
            }
            catch (Exception ex)
            {
                OutputDebugInfo(string.Format("斷開連接失敗：{0}", ex.ToString()));
            }
            OutputDebugInfo(string.Format("銷毀{0}", connectionInfo.No));
        }

        /// <summary>
        /// 發送
        /// </summary>
        /// <typeparam name="T1"></typeparam>
        /// <typeparam name="T2"></typeparam>
        /// <param name="rstData"></param>
        /// <param name="callback"></param>
        public void Send(byte[] rstData, Action<string, byte[]> callback)
        {
            var connInfo = GrabConnection();
            if (connInfo == null)
            {
                OutputDebugInfo("未獲取到可用連接");
                callback?.Invoke("無可用連接", null);
                return;
            }
            OutputDebugInfo(string.Format("獲取到連接{0}", connInfo.No));
            if (connInfo.Connection.IsConnected)
            {
                connInfo.LastUsedTime = DateTime.Now;
                try
                {
                    connInfo.Connection.Send(rstData,
                    (error, rndData) =>
                    {
                        callback?.BeginInvoke(error, rndData, null, null);
                        //重新加入空閑隊列
                        if (string.IsNullOrEmpty(error) && m_Connections.ContainsKey(connInfo.No))
                        {
                            m_IdleConnections.Enqueue(connInfo);
                        }
                    });
                }
                catch (Exception ex)
                {
                    OutputDebugInfo(string.Format("發送失敗：{0}", ex.ToString()));
                }
            }
            else
            {
                callback?.BeginInvoke("斷開連接", null, null, null);
                DestoryConnection(connInfo);
            }
        }

        /// <summary>
        /// 是否已清理
        /// </summary>
        private bool m_IsDisposed = false;
        /// <summary>
        /// 清理
        /// </summary>
        public void Dispose()
        {
            if (m_IsDisposed)
            {
                return;
            }
            m_IsDisposed = true;

            Clear(true);
            OutputDebugInfo("釋放完成");
        }

        private object m_ClearLock = new object();

        /// <summary>
        /// 開始清理
        /// </summary>
        private void StartClear()
        {
            ThreadPool.QueueUserWorkItem(
                (obj) =>
                {
                    Clear(false);
                    Thread.Sleep(1000);
                    if (!m_IsDisposed)
                    {
                        StartClear();
                    }
                });
        }

        /// <summary>
        /// 清理
        /// </summary>
        private void Clear(bool isForece)
        {
            lock (m_ClearLock)
            {
                var nos = m_Connections.Keys.ToList();
                ConnectionInfo connInfo = null;
                foreach (var no in nos)
                {
                    if (m_Connections.TryGetValue(no, out connInfo))
                    {
                        try
                        {
                            connInfo.Connection.Send(new byte[0], null);
                            IsConnected = true;
                        }
                        catch
                        {
                            IsConnected = false;
                        }
                        if ((m_Connections.Count > m_Min && DateTime.Now - connInfo.LastUsedTime > m_Expired) || isForece)
                        {
                            DestoryConnection(connInfo);
                        }
                    }
                }
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="debugInfo"></param>
        private void OutputDebugInfo(string debugInfo)
        {
            System.Diagnostics.Debug.WriteLine(string.Format("{0}-ConnectionPool-{1}", m_ConnectionFactory == null ? "Server" : "Client", debugInfo));
        }

        public event EventHandler Connected;
        public event EventHandler Disconnected;
    }