数据库连接池的计数器设计

2023-11-26 23:02:28

设计过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; }
    }
}

Counter完整代码

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