Thrift搭建分布式微服務（二）

第二篇 連接池

　　  連接池配置，請前往Thrift搭建分布式微服務（一）

　　  下面要介紹的其實不是單一的連接池，應該說是連接池集合。因為它要管理多個Tcp Socket連接節點，每個服務節點都有設置了自己的最大激活連接數、最大空閑連接數、最小空閑連接數、等待連接時間。

    internal class ServiceTransportPool
    {
        public ServiceConfig ServiceConfig { get; set; }

        public ConcurrentStack<TTransport> TransportPool { get; set; }

        public AutoResetEvent ResetEvent { get; set; }

        public int ActivedTransportCount { get; set; }
  　　　　　
　　　　    internal object sync_obeject = new object();

    }

　　

        一個ServiceTransportPool類對應一個服務配置，一個服務配置對應一個服務節點。連接池集合應具有下列成員：

    internal class ThriftFactory
    {
        private static volatile List<ServiceTransportPool> transport_pools;
        
        private static object sync_obj = new object();

        private static IThriftFactoryMonitor monitor = new ThriftFactoryMonitor();
    }

        transport_pools實現了對服務節點的管理，monitor 用來監控連接池的狀態，如上一篇所講，等待連接超時怎么通知連接池管理者，就用monitor來實現。這里monitor有個默認的實現，后面再講。

        初始化連接池集合：

        static ThriftFactory()
        {
            if (transport_pools == null || transport_pools.Count == 0)
            {
                lock (sync_obj)
                {
                    if (transport_pools == null || transport_pools.Count == 0)
                    {
                        var services = ConfigHelper.GetServiceConfigs();
                        transport_pools = new List<ServiceTransportPool>(services.Count);
                        foreach (var service in services)
                        {
                            ServiceTransportPool stp = new ServiceTransportPool()
                            {
                                ServiceConfig = service,
                                TransportPool = new ConcurrentStack<TTransport>(),
                                ResetEvent = new AutoResetEvent(false),
                                ActivedTransportCount = 0
                            };
                            transport_pools.Add(stp);
                        }
                    }
                }
            }
       }

 

　　 如何向連接池借出一個Socket連接：

        public static TTransport BorrowInstance(string serviceName)
        {
            var transpool = (from tp in transport_pools where tp.ServiceConfig.Name.ToUpper() == serviceName.ToUpper() select tp).FirstOrDefault();
            if (transpool == null)
            {
                throw new ThriftException(string.Format("There Is No Service Named \"{0}\"", serviceName));
            }

            TTransport transport;
            lock (transpool.sync_obeject)
            {
                if (transpool.TransportPool.Count() == 0)
                {
                    if (transpool.ActivedTransportCount == transpool.ServiceConfig.MaxActive)
                    {
                        bool result = transpool.ResetEvent.WaitOne(transpool.ServiceConfig.WaitingTimeout);
                        if (!result)
                        {
                            monitor.TimeoutNotify(transpool.ServiceConfig.Name, transpool.ServiceConfig.WaitingTimeout);
                        }
                    }
                    else
                    {
                        transpool.TransportPool.Push(CreateTransport(transpool.ServiceConfig));
                    }
                }

                if (!transpool.TransportPool.TryPop(out transport))
                {
                    throw new ThriftException("Connection Pool Exception");
                }

                transpool.ActivedTransportCount++;

                if (transpool.TransportPool.Count() < transpool.ServiceConfig.MinIdle && transpool.ActivedTransportCount < transpool.ServiceConfig.MaxActive)
                {
                    transpool.TransportPool.Push(CreateTransport(transpool.ServiceConfig));
                }
            }
            if (!transport.IsOpen)
            {
                transport.Open();
            }
            Monitor();
            return transport;
        }

        當實際激活的連接數達到服務節點配置的最大激活連接數，獲取Socket連接的請求就將處於等待狀態，超過等待時間設置，使用監視器方法monitor.TimeoutNotify()去通知管理者。連接池空閑的連接小於最小空閑連接數設置，每次請求連接都會建立一個新的連接放到池子里。

        歸還連接：

        public static void ReturnInstance(string serviceName,TTransport transport)
        {
            var transpool = (from tp in transport_pools where tp.ServiceConfig.Name.ToUpper() == serviceName.ToUpper() select tp).FirstOrDefault();
            if (transpool == null)
            {
                throw new ThriftException("Connection Pool Exception");
            }
            if (transpool.TransportPool.Count() == transpool.ServiceConfig.MaxIdle)
            {
                transport.Flush();
                if (transport.IsOpen)
                {
                    transport.Close();
                }
                transport.Dispose();
            }
            else
            {
                lock (transpool.sync_obeject)
                {
                    if (transport.IsOpen)
                    {
                        transport.Close();
                    }
                    transpool.TransportPool.Push(transport);
                    transpool.ActivedTransportCount--;
                    transpool.ResetEvent.Set();
                }
            }
            Monitor();
        }

        當連接池最大空閑連接達到了服務節點設置的最大空閑連接數時，歸還的連接將被銷毀。借出連接和歸還連接兩段代碼里都有一個Monitor()方法，此方法監控連接池連接的使用情況：

        /// <summary>
        /// 監控連接池狀態
        /// </summary>
        private static void Monitor()
        {
            List<Tuple<string, int, int>> tuples = new List<Tuple<string, int, int>>(transport_pools.Count);
            foreach(var transpool in transport_pools)
            {
                Tuple<string, int, int> tuple = new Tuple<string, int, int>(transpool.ServiceConfig.Name, transpool.TransportPool.Count(), transpool.ActivedTransportCount);
                tuples.Add(tuple);
            }
            monitor.Monitor(tuples);
        }

       此方法將每個服務連接池的空閑連接數、和激活的連接數傳給前面提到的監視器。在連接等待超時和拿到連接池的運行參數時，最終進行什么動作還是由開發者去實現的。繼承下面的接口，開發者可以自定義監視器。

    public interface IThriftFactoryMonitor
    {
        /// <summary>
        /// 監控連接池運行狀態
        /// </summary>
        /// <param name="tuple">元組集合，第一個元素表示服務名稱、第二個元素表示空閑連接數量、第三個元素表示激活連接數量</param>
        void Monitor(List<Tuple<string,int,int>> tuples);

        /// <summary>
        /// 等待連接超時
        /// </summary>
        void TimeoutNotify(string serviceName,int timeOut);
    }

       默認的監視器只是將連接池的運行狀態記錄到控制台：

    /// <summary>
    /// 默認連接池狀態監控類
    /// </summary>
    public class ThriftFactoryMonitor : IThriftFactoryMonitor
    {
        public virtual void Monitor(List<Tuple<string, int, int>> tuples)
        {
            foreach (var t in tuples)
            {
                Console.WriteLine(string.Format("{0}連接池，空閑連接數量：{1}，激活連接數量：{2}", t.Item1, t.Item2, t.Item3));
            }
        }

        public virtual void TimeoutNotify(string serviceName, int timeOut)
        {
            Console.WriteLine(string.Format("{0}連接池等待連接超時{1}", serviceName, timeOut));
        }
    }

　　  開發者自己實現的監視器，如何能被連接池使用，其實在一開始的連接池初始化里，還有一段使用反射來初始化開發者定義的監視器的代碼。開發者只需在上一篇介紹的Thrift.config里配置MonitorType，其他的就交給框架處理：

        static ThriftFactory()
        {
            ......
            if(!string.IsNullOrWhiteSpace(ConfigHelper.ThriftConfig.MonitorType))
            {
                monitor = Invoker.CreateInstance(Type.GetType(ConfigHelper.ThriftConfig.MonitorType)) as IThriftFactoryMonitor;
                if (monitor == null)
                {
                    throw new ThriftException(string.Format("There Is No Monitor Implement Which Type Is  \"{0}\"", ConfigHelper.ThriftConfig.MonitorType));
                }
            }
        }

　　 通過連接池與服務節點建立了Socket連接，下一篇將介紹客戶端如何使用建立的Socket連接與服務端通信。

Thrift微服務代碼下載Thrift.Utility