數據庫連接池的計數器設計

設計過ORM的攻城獅們或多或少都應該考慮過連接池,考慮過連接池就或多或少會想過計數器....

<計數器在連接池中的應用>

曾經在我設計一套ORM的時候想過這樣一種連接池的管理方式:

• 0.連接池中的所有連接都對應存在一個計數器,指示連接目前被多少對象引用;
  當計數器從0變成1的時候,打開連接Connection.Open();
  當計數器從1變成0的時候,關閉連接Connection.Close();
• 1.連接池中有一個默認連接DefaultConnection,這個連接被所有的非事務操作共用,比如查(select);
• 2.當發生一個查詢操作的時候,先獲得DefaultConnection,同時對應計數器+1,使用完之后DefaultConnection的計數器-1;
• 3.當發生事務操作的時候,會從連接池中申請一個連接數為0的Connection(但不會是DefaultConnection);
  如果連接池中不存在這樣的連接,則會新建一個並加入到連接池中;
  獲得Connection后對應計數器+1,使用完之后對應計數器-1;
• 4.如果申請事務操作時連接池已達到上限,且所有連接的計數器都大於1,則請求進入隊列,直至得到Connection或超時;

<計數器1.0>

第一版的設計非常的簡單,直接就是類似於這樣的
ps:以下為示例代碼,用意是便於理解,請不要太較真

class MyConnection
{
    public IDbConnection Connection { get; private set; }

    int _linkedCount;
    public void Add()
    {
        var i = Interlocked.Increment(ref _linkedCount);
        if (i == 1)
        {
            Connection.Open();
        }
    }

    public void Remove()
    {
        var i = Interlocked.Decrement(ref _linkedCount);
        if (i == 0)
        {
            Connection.Close();
        }
    }
}

class ORM
{
    public MyConnection Connection { get; private set; }

    public int ExecuteNonQuery(string sql)
    {
        try
        {
            Connection.Add();
            var cmd = Connection.Connection.CreateCommand();
            cmd.CommandText = sql;
            return cmd.ExecuteNonQuery();
        }
        finally
        {
            Connection.Remove();
        }
    }
}

使用

using (ORM db = new ORM())
{
    db.ExecuteNonQuery("insert xxx,xxx,xx");
}

<設計缺陷>

但是緊接着就出現一個問題了
如果我有一個方法,需要同時進行多個操作
比如

using (ORM db = new ORM())
{
    db.ExecuteNonQuery("insert aaa");
    db.ExecuteNonQuery("insert bbb");
    db.ExecuteNonQuery("insert ccc");
    db.ExecuteNonQuery("insert ddd");
}

這樣其實已經開啟關閉了4次數據庫

這樣的性能損耗是非常大的

所以我考慮這樣的模式

using (ORM db = new ORM())
{
    db.Open();
    db.ExecuteNonQuery("insert aaa");
    db.ExecuteNonQuery("insert bbb");
    db.ExecuteNonQuery("insert ccc");
    db.ExecuteNonQuery("insert ddd");
    db.Close();
}

這樣有經驗的朋友一眼就可以看出更大的問題

如果insert ddd的報錯了怎么辦 Close就無法關閉了
換一種方式說,如果coder忘記寫Close(),或者某個分支中忘記寫Close()怎么辦?

難道我要求所有coder都要寫try..finally..?
也許你會說把Close放到using的Dispose方法中去

class ORM : IDisposable
{
    public MyConnection Connection { get; private set; }

    public int ExecuteNonQuery(string sql)
    {
        try
        {
            Connection.Add();
            var cmd = Connection.Connection.CreateCommand();
            cmd.CommandText = sql;
            return cmd.ExecuteNonQuery();
        }
        finally
        {
            Connection.Remove();
        }
    }
    
    public void Open()
    {
        Connection.Add();
    }

    public void Close()
    {
        Connection.Remove();
    }
    public void Dispose()
    {
        Close();
    }
}

但是,如果這樣 coder已經寫了Close() 或者根本沒寫Open() 不是會多觸發一個Remove()?

那豈不是會出現計數器=-1,-2...-N

<計數器 N.0>

其實我也不記得我嘗試過多少種方案了,我只記得最終我是這樣實現我想要的效果的:

• 0.首先,每個Add()加增的計數只有對應的Remove()可以減少

  為了實現這一目標,每個Add()將會返回一個對象,而Remove(token)將接受這個對象,以便於控制-1這樣的操作;
  

  var token = Connection.Add();    //計數器+1
  ...
  ...
  Connection.Remove(token);        //計數器-1
  Connection.Remove(token);        //無效果
  Connection.Remove(token);        //無效果

  為了更加優化這樣的效果,我將Add()的返回值設置為IDisposable
  也就是說可以這樣寫

  using (Connection.Add())
  {
      //...
      //...
  }

  或者這樣寫

  var token = Connection.Add();
  //...
  //...
  token.Dispose();

• 1.在同一個線程中,只有第一次執行Add會讓計數器增加,同樣,只有第一次執行Add的返回對象可以減少計數器;

  var token1 = Connection.Add();    //計數器+1
  var token2 = Connection.Add();    //無效果
  var token3 = Connection.Add();    //無效果
  //...
  //...
  Connection.Remove(token3);        //無效果
  Connection.Remove(token2);        //無效果
  Connection.Remove(token1);        //計數器-1

  需要實現這個效果,就必須利用LocalDataStoreSlot對象

  /// <summary> 用於儲存多線程間的獨立數據
  /// </summary>
  private LocalDataStoreSlot _dataSlot = Thread.AllocateDataSlot();
  
  /// <summary> 增加引用,並獲取用於釋放引用的標記
  /// </summary>
  public IDisposable Add()
  {
      //如果已經存在,則不計數
      if (Thread.GetData(_dataSlot) != null)//如果變量值已經存在,則說明當前線程已經執行Add方法,則返回null
      {
          return null;
      }
      Thread.SetData(_dataSlot, string.Empty);//在當前線程中保存一個變量值
      return new CounterToken(this);
  }
  
  /// <summary> 減少引用
  /// </summary>
  /// <param name="token">通過Add方法獲取的標記對象</param>
  public void Reomve(IDisposable token)
  {
      if (token == null)
      {
          return;
      }
      if (token is CounterToken == false)
      {
          throw new ArgumentException("參數不是一個有效的引用標記", "token");
      }
      if (token.Equals(this) == false)//CounterToken已經重寫Equals方法
      {
          throw new ArgumentOutOfRangeException("token", "此標記不屬於當前計數器");
      }
      token.Dispose();
  }

  其中CounterToken就是計數器的標記,實現IDisposable接口,是一個內部類

<問題解決>

通過這樣2部步設置,就可以實現之前無法完成的效果了

而ORM部分的代碼需要稍微修改下

class ORM : IDisposable
{
    public MyConnection Connection { get; private set; }

    public int ExecuteNonQuery(string sql)
    {
        //try
        //{
            //Connection.Add();
        using (Connection.Add())
        {
            var cmd = Connection.Connection.CreateCommand();
            cmd.CommandText = sql;
            return cmd.ExecuteNonQuery();
        }
        //}
        //finally
        //{
        //    Connection.Remove();
        //}
        //return -1;
    }

    IDisposable _counterToken;

    public void Open()
    {
        if (_counterToken == null)
        {
            _counterToken = Connection.Add();
        }
    }

    public void Close()
    {
        Connection.Remove(_counterToken);
        _counterToken = null;
    }

    public void Dispose()
    {
        Close();
        Connection = null;
    }
}

調用的時候

using (ORM db = new ORM())
{
    db.Open();
    db.ExecuteNonQuery("insert aaa");
    db.ExecuteNonQuery("insert bbb");
    db.ExecuteNonQuery("insert ccc");
    db.ExecuteNonQuery("insert ddd");
    db.Close();
}

完全沒有問題,只有一個Open()會增加計數器,最后一個Close()會減少計數器(如果有必要的話,他們會自動打開和關閉Connection());

關鍵的是,這樣做我得到了一個額外的好處;

即使coder即忘記了using,也忘記了Close...

沒關系,因為GC的存在,一旦CounterToken沒有被任何人應用而釋放掉了,那么計數器仍然會將他減掉;

<最后的封裝>

最后的最后,我把這個計數器從MyConection中獨立出來了(其實根本就不存在什么MyConection,都是我瞎編的,只是這樣說比較好理解而已,哈哈~~)

計數器分為2個模式 ,之前文章中介紹的都是多線程模式,單線程模式只是附帶的一個功能而已

單線程模式:無論在任何線程中每次執行Add方法都會增加引用數,執行Remove或者token.Dispose都會減少引用數

多線程模式:在相同線程中,只有第一次執行Add方法時增加引用數,也只有此token被Remove或Dispose才會減少引用數

ps:為了使計數器和數據庫組件解耦,所以我在計數器中設計了一個ValueChaged事件

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace blqw
{
    /// <summary> 計數器,具有單線程模式和多線程模式
    /// </summary>
    public sealed class Counter
    {
        /// <summary> 構造一個計數器,默認單線程模式
        /// <para>無論在任何線程中每次執行Add方法都會增加引用數,執行Remove或者token.Dispose都會減少引用數</para>
        /// </summary>
        public Counter()
            :this(false)
        {
            Console.WriteLine();
        }
        /// <summary> 構造一個計數器,根據參數multiThreadMode確定是否使用多線程模式
        /// <para>多線程模式:在相同線程中,只有第一次執行Add方法時增加引用數,也只有此token被Remove或Dispose才會減少引用數</para>
        /// </summary>
        /// <param name="multiThreadMode"></param>
        public Counter(bool multiThreadMode)
        {
            if (multiThreadMode)
            {
                _dataSlot = Thread.AllocateDataSlot();
            }
        }

        /// <summary> 當前引用數
        /// </summary>
        private int _value; 
        /// <summary> 值改變事件
        /// </summary>
        private EventHandler<CounterChangedEventArgs> _valueChanged;
        /// <summary> 用於儲存多線程間的獨立數據,多線程模式下有值
        /// </summary>
        private LocalDataStoreSlot _dataSlot;

        /// <summary> 增加引用,並獲取用於釋放引用的標記
        /// </summary>
        public IDisposable Add()
        {
            if (_dataSlot != null)
            {
                //獲取當前線程中的值,此方法每個線程中獲得的值都不同,不需要線程同步
                //如果已經存在,則不計數
                if (Thread.GetData(_dataSlot) != null)
                {
                    return null;
                }
                Thread.SetData(_dataSlot, string.Empty);
            }
            return new CounterToken(this);
        }

        /// <summary> 減少引用
        /// </summary>
        /// <param name="token">通過Add方法獲取的標記對象</param>
        public void Remove(IDisposable token)
        {
            if (token == null)
            {
                return;
            }
            if (token is CounterToken == false)
            {
                throw new ArgumentException("參數不是一個有效的引用標記", "token");
            }
            if (token.Equals(this) == false)
            {
                throw new ArgumentOutOfRangeException("token", "此標記不屬於當前計數器");
            }
            token.Dispose();
        }

        /// <summary> 當前計數值
        /// </summary>
        public int Value
        {
            get { return _value; }
        }

        /// <summary> 增加記數
        /// </summary>
        private void OnIncrement()
        {
            var val = Interlocked.Increment(ref _value);
            OnValueChanged(val, val - 1);
        }
        /// <summary> 減少計數
        /// </summary>
        private void OnDecrement()
        {
            if (_dataSlot != null)
            {
                Thread.SetData(_dataSlot, null);
            }
            var val = Interlocked.Decrement(ref _value);
            OnValueChanged(val, val + 1);
        }
        /// <summary> 觸發ValueChaged事件
        /// </summary>
        /// <param name="value">觸發Value事件時Value的值</param>
        /// <param name="oldValue">觸發Value事件之前Value的值</param>
        private void OnValueChanged(int value, int oldValue)
        {
            var handler = _valueChanged;
            if (handler != null)
            {
                var e = new CounterChangedEventArgs(value, oldValue);
                handler(this, e);
            }
        }
        /// <summary> 計數器值改變事件
        /// </summary>
        public event EventHandler<CounterChangedEventArgs> ValueChanged
        {
            add
            {
                _valueChanged -= value;
                _valueChanged += value;
            }
            remove
            {
                _valueChanged -= value;
            }
        }

        /// <summary> 計數器引用標記,調用計數器的Add方法可獲得該對象,釋放對象時,減少計數器的計數值
        /// </summary>
        sealed class CounterToken : IDisposable
        {
            /// <summary> 宿主計數器
            /// </summary>
            private Counter _counter;
            /// <summary> 釋放標記,0未釋放,1已釋放,2執行了析構函數
            /// </summary>
            private int _disposeMark;
            /// <summary> 構造函數,創建引用標記並增加宿主計數器的值
            /// </summary>
            /// <param name="counter">宿主計數器</param>
            public CounterToken(Counter counter)
            {
                if (counter == null)
                {
                    throw new ArgumentNullException("counter");
                }
                _counter = counter;
                _counter.OnIncrement();
                _disposeMark = 0;
            }
            /// <summary> 析構函數
            /// </summary>
            ~CounterToken()
            {
                //如果尚未釋放對象(標記為0),則將標記改為2,否則標記不變
                Interlocked.CompareExchange(ref _disposeMark, 2, 0);
                Dispose();
            }
            /// <summary> 釋放引用標記,並減少宿主計數器的值
            /// </summary>
            public void Dispose()
            {
                //如果已釋放(標記為1)則不執行任何操作
                if (_disposeMark == 1)
                {
                    return;
                }
                //將標記改為1,並返回修改之前的值
                var mark = Interlocked.Exchange(ref _disposeMark, 1);
                //如果當前方法被多個線程同時執行,確保僅執行其中的一個
                if (mark == 1)
                {
                    return;
                }
                //釋放Counter引用數
                try
                {
                    _counter.OnDecrement();
                }
                catch
                {
                    
                }
                _counter = null;
                //如果mark=0,則通知系統不需要執行析構函數了
                if (mark == 0)
                {
                    GC.SuppressFinalize(this);
                }
            }
            /// <summary> 重新實現比較的方法
            /// </summary>
            /// <param name="obj"></param>
            /// <returns></returns>
            public override bool Equals(object obj)
            {
                if (obj is Counter)
                {
                    return object.ReferenceEquals(this._counter, obj);
                }
                return object.ReferenceEquals(this, obj);
            }
            
        }
    }

    /// <summary> 計數器值改變事件的參數
    /// </summary>
    public class CounterChangedEventArgs:EventArgs
    {

        internal CounterChangedEventArgs(int value,int oldValue)
        {
            Value = value;
            OldValue = oldValue;
        }
        /// <summary> 當前值
        /// </summary>
        public int Value { get; private set; }
        /// <summary> 原值
        /// </summary>
        public int OldValue { get; private set; }
    }
}

 

var counter = new Counter(true);//多線程模式
//var counter = new Counter();  //單線程模式

new Thread(() =>
{
    using (counter.Add())           //計數器+1  當前計數器=1
    {
        Console.WriteLine("線程a:" + counter.Value);
        using (counter.Add())       //計數器不變 當前計數器=1
        {
            Console.WriteLine("線程a:" + counter.Value);
            using (counter.Add())   //計數器不變 當前計數器=1
            {
                Console.WriteLine("線程a:" + counter.Value);
                Thread.Sleep(100);  //等待線程b執行,b執行完之后 當前計數器=1
            }                       //計數器不變 當前計數器=1
            Console.WriteLine("線程a:" + counter.Value);
        }                           //計數器不變 當前計數器=1
        Console.WriteLine("線程a:" + counter.Value);
    }                               //計數器-1  當前計數器=0
    Console.WriteLine("線程a:" + counter.Value);
}).Start();

Thread.Sleep(50);
new Thread(() =>
{
    var token1 = counter.Add();    //計數器+1  當前計數器=2
    Console.WriteLine("線程b:" + counter.Value);
    var token2 = counter.Add();    //計數器不變 當前計數器=2
    Console.WriteLine("線程b:" + counter.Value);
    var token3 = counter.Add();    //計數器不變 當前計數器=2
    Console.WriteLine("線程b:" + counter.Value);
    counter.Remove(token3);        //計數器不變 當前計數器=2
    Console.WriteLine("線程b:" + counter.Value);
    counter.Remove(token2);        //計數器不變 當前計數器=2
    Console.WriteLine("線程b:" + counter.Value);
    counter.Remove(token1);        //計數器-1  當前計數器=1
    Console.WriteLine("線程b:" + counter.Value);
}).Start();
Console.ReadLine();

 

多線程模式測試結果

 

單線程模式測試結果