數(shù)據(jù)庫(kù)連接池的計(jì)數(shù)器設(shè)計(jì)

2019-11-14 16:40:18

字體：大中小

來源：轉(zhuǎn)載

供稿：網(wǎng)友

設(shè)計(jì)過ORM的攻城獅們或多或少都應(yīng)該考慮過連接池,考慮過連接池就或多或少會(huì)想過計(jì)數(shù)器....

<計(jì)數(shù)器在連接池中的應(yīng)用>

曾經(jīng)在我設(shè)計(jì)一套ORM的時(shí)候想過這樣一種連接池的管理方式:

0.連接池中的所有連接都對(duì)應(yīng)存在一個(gè)計(jì)數(shù)器,指示連接目前被多少對(duì)象引用;
當(dāng)計(jì)數(shù)器從0變成1的時(shí)候,打開連接Connection.Open();
當(dāng)計(jì)數(shù)器從1變成0的時(shí)候,關(guān)閉連接Connection.Close();
1.連接池中有一個(gè)默認(rèn)連接DefaultConnection,這個(gè)連接被所有的非事務(wù)操作共用,比如查(select);
2.當(dāng)發(fā)生一個(gè)查詢操作的時(shí)候,先獲得DefaultConnection,同時(shí)對(duì)應(yīng)計(jì)數(shù)器+1,使用完之后DefaultConnection的計(jì)數(shù)器-1;
3.當(dāng)發(fā)生事務(wù)操作的時(shí)候,會(huì)從連接池中申請(qǐng)一個(gè)連接數(shù)為0的Connection(但不會(huì)是DefaultConnection);
如果連接池中不存在這樣的連接,則會(huì)新建一個(gè)并加入到連接池中;
獲得Connection后對(duì)應(yīng)計(jì)數(shù)器+1,使用完之后對(duì)應(yīng)計(jì)數(shù)器-1;
4.如果申請(qǐng)事務(wù)操作時(shí)連接池已達(dá)到上限,且所有連接的計(jì)數(shù)器都大于1,則請(qǐng)求進(jìn)入隊(duì)列,直至得到Connection或超時(shí);

<計(jì)數(shù)器1.0>

第一版的設(shè)計(jì)非常的簡(jiǎn)單,直接就是類似于這樣的
ps:以下為示例代碼,用意是便于理解,請(qǐng)不要太較真

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");}

<設(shè)計(jì)缺陷>

但是緊接著就出現(xiàn)一個(gè)問題了
如果我有一個(gè)方法,需要同時(shí)進(jìn)行多個(gè)操作
比如

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

這樣其實(shí)已經(jīng)開啟關(guān)閉了4次數(shù)據(jù)庫(kù)

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

所以我考慮這樣的模式

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();}

這樣有經(jīng)驗(yàn)的朋友一眼就可以看出更大的問題

如果insert ddd的報(bào)錯(cuò)了怎么辦 Close就無法關(guān)閉了
換一種方式說,如果coder忘記寫Close(),或者某個(gè)分支中忘記寫Close()怎么辦?

難道我要求所有coder都要寫try..finally..?
也許你會(huì)說把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已經(jīng)寫了Close() 或者根本沒寫Open() 不是會(huì)多觸發(fā)一個(gè)Remove()?

那豈不是會(huì)出現(xiàn)計(jì)數(shù)器=-1,-2...-N

<計(jì)數(shù)器 N.0>

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

0.首先,每個(gè)Add()加增的計(jì)數(shù)只有對(duì)應(yīng)的Remove()可以減少
為了實(shí)現(xiàn)這一目標(biāo),每個(gè)Add()將會(huì)返回一個(gè)對(duì)象,而Remove(token)將接受這個(gè)對(duì)象,以便于控制-1這樣的操作;
```
var token = Connection.Add();    //計(jì)數(shù)器+1......Connection.Remove(token);        //計(jì)數(shù)器-1Connection.Remove(token);        //無效果Connection.Remove(token);        //無效果
```
為了更加優(yōu)化這樣的效果,我將Add()的返回值設(shè)置為IDisposable
也就是說可以這樣寫
```
using (Connection.Add()){    //...    //...}
```
或者這樣寫
```
var token = Connection.Add();//...//...token.Dispose();
```

1.在同一個(gè)線程中,只有第一次執(zhí)行Add會(huì)讓計(jì)數(shù)器增加,同樣,只有第一次執(zhí)行Add的返回對(duì)象可以減少計(jì)數(shù)器;

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

需要實(shí)現(xiàn)這個(gè)效果,就必須利用LocalDataStoreSlot對(duì)象

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

其中CounterToken就是計(jì)數(shù)器的標(biāo)記,實(shí)現(xiàn)IDisposable接口,是一個(gè)內(nèi)部類

<問題解決>

通過這樣2部步設(shè)置,就可以實(shí)現(xiàn)之前無法完成的效果了

而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;    }}

調(diào)用的時(shí)候

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();}

完全沒有問題,只有一個(gè)Open()會(huì)增加計(jì)數(shù)器,最后一個(gè)Close()會(huì)減少計(jì)數(shù)器(如果有必要的話,他們會(huì)自動(dòng)打開和關(guān)閉Connection());

關(guān)鍵的是,這樣做我得到了一個(gè)額外的好處;

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

沒關(guān)系,因?yàn)镚C的存在,一旦CounterToken沒有被任何人應(yīng)用而釋放掉了,那么計(jì)數(shù)器仍然會(huì)將他減掉;

<最后的封裝>

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

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

單線程模式:無論在任何線程中每次執(zhí)行Add方法都會(huì)增加引用數(shù),執(zhí)行Remove或者token.Dispose都會(huì)減少引用數(shù)
多線程模式:在相同線程中,只有第一次執(zhí)行Add方法時(shí)增加引用數(shù),也只有此token被Remove或Dispose才會(huì)減少引用數(shù)

ps:為了使計(jì)數(shù)器和數(shù)據(jù)庫(kù)組件解耦,所以我在計(jì)數(shù)器中設(shè)計(jì)了一個(gè)ValueChaged事件

using System;using System.Collections.Generic;using System.Text;using System.Threading;namespace blqw{    /// <summary> 計(jì)數(shù)器,具有單線程模式和多線程模式    /// </summary>    public sealed class Counter    {        /// <summary> 構(gòu)造一個(gè)計(jì)數(shù)器,默認(rèn)單線程模式        /// <para>無論在任何線程中每次執(zhí)行Add方法都會(huì)增加引用數(shù),執(zhí)行Remove或者token.Dispose都會(huì)減少引用數(shù)</para>        /// </summary>        public Counter()            :this(false)        {            Console.WriteLine();        }        /// <summary> 構(gòu)造一個(gè)計(jì)數(shù)器,根據(jù)參數(shù)multiThreadMode確定是否使用多線程模式        /// <para>多線程模式:在相同線程中,只有第一次執(zhí)行Add方法時(shí)增加引用數(shù),也只有此token被Remove或Dispose才會(huì)減少引用數(shù)</para>        /// </summary>        /// <param name="multiThreadMode"></param>        public Counter(bool multiThreadMode)        {            if (multiThreadMode)            {                _dataSlot = Thread.AllocateDataSlot();            }        }        /// <summary> 當(dāng)前引用數(shù)        /// </summary>        private int _value;         /// <summary> 值改變事件        /// </summary>        private EventHandler<CounterChangedEventArgs> _valueChanged;        /// <summary> 用于儲(chǔ)存多線程間的獨(dú)立數(shù)據(jù),多線程模式下有值        /// </summary>        private LocalDataStoreSlot _dataSlot;        /// <summary> 增加引用,并獲取用于釋放引用的標(biāo)記        /// </summary>        public IDisposable Add()        {            if (_dataSlot != null)            {                //獲取當(dāng)前線程中的值,此方法每個(gè)線程中獲得的值都不同,不需要線程同步                //如果已經(jīng)存在,則不計(jì)數(shù)                if (Thread.GetData(_dataSlot) != null)                {                    return null;                }                Thread.SetData(_dataSlot, string.Empty);            }            return new CounterToken(this);        }        /// <summary> 減少引用        /// </summary>        /// <param name="token">通過Add方法獲取的標(biāo)記對(duì)象</param>        public void Remove(IDisposable token)        {            if (token == null)            {                return;            }            if (token is CounterToken == false)            {                throw new ArgumentException("參數(shù)不是一個(gè)有效的引用標(biāo)記", "token");            }            if (token.Equals(this) == false)            {                throw new ArgumentOutOfRangeException("token", "此標(biāo)記不屬于當(dāng)前計(jì)數(shù)器");            }            token.Dispose();        }        /// <summary> 當(dāng)前計(jì)數(shù)值        /// </summary>        public int Value        {            get { return _value; }        }        /// <summary> 增加記數(shù)        /// </summary>        private void OnIncrement()        {            var val = Interlocked.Increment(ref _value);            OnValueChanged(val, val - 1);        }        /// <summary> 減少計(jì)數(shù)        /// </summary>        private void OnDecrement()        {            if (_dataSlot != null)            {                Thread.SetData(_dataSlot, null);            }            var val = Interlocked.Decrement(ref _value);            OnValueChanged(val, val + 1);        }        /// <summary> 觸發(fā)ValueChaged事件        /// </summary>        /// <param name="value">觸發(fā)Value事件時(shí)Value的值</param>        /// <param name="oldValue">觸發(fā)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> 計(jì)數(shù)器值改變事件        /// </summary>        public event EventHandler<CounterChangedEventArgs> ValueChanged        {            add            {                _valueChanged -= value;                _valueChanged += value;            }            remove            {                _valueChanged -= value;            }        }        /// <summary> 計(jì)數(shù)器引用標(biāo)記,調(diào)用計(jì)數(shù)器的Add方法可獲得該對(duì)象,釋放對(duì)象時(shí),減少計(jì)數(shù)器的計(jì)數(shù)值        /// </summary>        sealed class CounterToken : IDisposable        {            /// <summary> 宿主計(jì)數(shù)器            /// </summary>            private Counter _counter;            /// <summary> 釋放標(biāo)記,0未釋放,1已釋放,2執(zhí)行了析構(gòu)函數(shù)            /// </summary>            private int _disposeMark;            /// <summary> 構(gòu)造函數(shù),創(chuàng)建引用標(biāo)記并增加宿主計(jì)數(shù)器的值            /// </summary>            /// <param name="counter">宿主計(jì)數(shù)器</param>            public CounterToken(Counter counter)            {                if (counter == null)                {                    throw new ArgumentNullException("counter");                }                _counter = counter;                _counter.OnIncrement();                _disposeMark = 0;            }            /// <summary> 析構(gòu)函數(shù)            /// </summary>            ~CounterToken()            {                //如果尚未釋放對(duì)象(標(biāo)記為0),則將標(biāo)記改為2,否則標(biāo)記不變                Interlocked.CompareExchange(ref _disposeMark, 2, 0);                Dispose();            }            /// <summary> 釋放引用標(biāo)記,并減少宿主計(jì)數(shù)器的值            /// </summary>            public void Dispose()            {                //如果已釋放(標(biāo)記為1)則不執(zhí)行任何操作                if (_disposeMark == 1)                {                    return;                }                //將標(biāo)記改為1,并返回修改之前的值                var mark = Interlocked.Exchange(ref _disposeMark, 1);                //如果當(dāng)前方法被多個(gè)線程同時(shí)執(zhí)行,確保僅執(zhí)行其中的一個(gè)                if (mark == 1)                {                    return;                }                //釋放Counter引用數(shù)                try                {                    _counter.OnDecrement();                }                catch                {                                    }                _counter = null;                //如果mark=0,則通知系統(tǒng)不需要執(zhí)行析構(gòu)函數(shù)了                if (mark == 0)                {                    GC.SuppressFinalize(this);                }            }            /// <summary> 重新實(shí)現(xiàn)比較的方法            /// </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> 計(jì)數(shù)器值改變事件的參數(shù)    /// </summary>    public class CounterChangedEventArgs:EventArgs    {        internal CounterChangedEventArgs(int value,int oldValue)        {            Value = value;            OldValue = oldValue;        }        /// <summary> 當(dāng)前值        /// </summary>        public int Value { get; private set; }        /// <summary> 原值        /// </summary>        public int OldValue { get; private set; }    }}

Counter完整代碼

var counter = new Counter(true);//多線程模式//var counter = new Counter();  //單線程模式new Thread(() =>{    using (counter.Add())           //計(jì)數(shù)器+1  當(dāng)前計(jì)數(shù)器=1    {        Console.WriteLine("線程a:" + counter.Value);        using (counter.Add())       //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=1        {            Console.WriteLine("線程a:" + counter.Value);            using (counter.Add())   //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=1            {                Console.WriteLine("線程a:" + counter.Value);                Thread.Sleep(100);  //等待線程b執(zhí)行,b執(zhí)行完之后 當(dāng)前計(jì)數(shù)器=1            }                       //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=1            Console.WriteLine("線程a:" + counter.Value);        }                           //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=1        Console.WriteLine("線程a:" + counter.Value);    }                               //計(jì)數(shù)器-1  當(dāng)前計(jì)數(shù)器=0    Console.WriteLine("線程a:" + counter.Value);}).Start();Thread.Sleep(50);new Thread(() =>{    var token1 = counter.Add();    //計(jì)數(shù)器+1  當(dāng)前計(jì)數(shù)器=2    Console.WriteLine("線程b:" + counter.Value);    var token2 = counter.Add();    //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=2    Console.WriteLine("線程b:" + counter.Value);    var token3 = counter.Add();    //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=2    Console.WriteLine("線程b:" + counter.Value);    counter.Remove(token3);        //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=2    Console.WriteLine("線程b:" + counter.Value);    counter.Remove(token2);        //計(jì)數(shù)器不變 當(dāng)前計(jì)數(shù)器=2    Console.WriteLine("線程b:" + counter.Value);    counter.Remove(token1);        //計(jì)數(shù)器-1  當(dāng)前計(jì)數(shù)器=1    Console.WriteLine("線程b:" + counter.Value);}).Start();Console.ReadLine();