利用双缓冲队列来减少锁的竞争

Posted 2020-10-21 网络蚂蚁

　　在日常的开发中，日志的记录是必不可少的。但是我们也清楚对同一个文本进行写日志只能单线程的去写，那么我们也经常会使用简单lock锁来保证只有一个线程来写入日志信息。但是在多线程的去写日志信息的时候，由于记录日志信息是需要进行I/O交互的，导致我们占用锁的时间会加长，从而导致大量线程的阻塞与等待。

　　这种场景下我们就会去思考，我们该怎么做才能保证当有多个线程来写日志的时候我们能够在不利用锁的情况下让他们依次排队去写呢？这个时候我们就可以考虑下使用双缓冲队列来完成。

　　所谓双缓冲队列就是有两个队列，一个是用来专门负责数据的写入，另一个是专门负责数据的读取，当逻辑线程读取完数据后负责将自己的队列与I/O线程的队列进行交换。

　　我们该怎么利用这双缓冲队列来完成我们想要的效果呢？

　　当有多个线程来写日志的时候，这个时候我们要这些要写的信息先放到我们负责写入的队列当中，然后将逻辑读的线程设为非阻塞。此时逻辑读的线程就可以开始工作了。（一开始时逻辑读的队列是空的）在当逻辑读的线程读取他自己队列的数据（并执行一些逻辑）之后，将逻辑读的队列的引用和负责写入的队列进行引用交换。这就是简单的一个双缓冲队列实现的一个思路。具体实现代码如下：

　　

    public class User {
        public string Mobile { get; set; }

        public string Pwd { get; set; }

        public override string ToString() {
            return $"{Mobile},{Pwd}";
        }
    }

  

    public class DoubleQueue {
        private ConcurrentQueue<User> _writeQueue;
        private ConcurrentQueue<User> _readQueue;
        private volatile ConcurrentQueue<User> _currentQueue;

        private AutoResetEvent _dataEvent;
        private ManualResetEvent _finishedEvent;
        private ManualResetEvent _producerEvent;

        public DoubleQueue() {
            _writeQueue = new ConcurrentQueue<User>();
            _readQueue = new ConcurrentQueue<User>();
            _currentQueue = _writeQueue;

            _dataEvent = new AutoResetEvent(false);
            _finishedEvent = new ManualResetEvent(true);
            _producerEvent = new ManualResetEvent(true);
            Task.Factory.StartNew(() => ConsumerQueue(), TaskCreationOptions.None);
        }

        public void ProducerFunc(User user) {
            _producerEvent.WaitOne();
            _finishedEvent.Reset();
            _currentQueue.Enqueue(user);
            _dataEvent.Set();
            _finishedEvent.Set();
        }

        public void ConsumerQueue() {
            ConcurrentQueue<User> consumerQueue;
            User user;
            int allcount = 0;
            Stopwatch watch = Stopwatch.StartNew();
            while (true)
            {
                _dataEvent.WaitOne();
                if (_currentQueue.Count > 0)
                {
                    _producerEvent.Reset();
                    _finishedEvent.WaitOne();
                    consumerQueue = _currentQueue;
                    _currentQueue = (_currentQueue == _writeQueue) ? _readQueue : _writeQueue;
                    _producerEvent.Set();
                    while (consumerQueue.Count > 0)
                    {
                        if (consumerQueue.TryDequeue(out user))
                        {
                            FluentConsole.White.Background.Red.Line(user.ToString());
                            allcount++;
                        }
                        FluentConsole.White.Background.Red.Line($"当前个数{allcount.ToString()}，花费了{watch.ElapsedMilliseconds.ToString()}ms;");
                        System.Threading.Thread.Sleep(20);
                    }
                }
            }
        }
    }

FluentConsole 是一个控制台应用程序的输出插件，开源的，有兴趣的可以自己去玩玩。

    internal class Program {
        private static object obj = new object();

        private static void Main(string[] args) {
            DoubleQueue doubleQueue = new DoubleQueue();
            Parallel.For(0, 3000, i =>
            {
                User user = new User()
                {
                    Mobile = i.ToString().PadLeft(11, \'0\'),
                    Pwd = i.ToString().PadLeft(8, \'8\')
                };
                doubleQueue.ProducerFunc(user);
            });

            Stopwatch watch = Stopwatch.StartNew();
            int allcount = 0;
            Parallel.For(0, 3000, i =>
            {
                User user = new User()
                {
                    Mobile = i.ToString().PadLeft(11, \'0\'),
                    Pwd = i.ToString().PadLeft(8, \'8\')
                };
                lock (obj)
                {
                    FluentConsole.White.Background.Red.Line(user.ToString());
                    allcount++;
                    FluentConsole.White.Background.Red.Line($"当前个数{allcount.ToString()}，花费了{watch.ElapsedMilliseconds.ToString()}ms;");
                    System.Threading.Thread.Sleep(20);
                }
            });
            FluentConsole.Black.Background.Red.Line("执行完成");
            Console.Read();
        }
    }

第一个利用双缓冲队列来执行，第二个利用lock锁来执行。下面分别是第一种方法和第二种方法执行时CPU的消耗。

我们可以发现利用双队列缓冲的情况下我们减少了CPU的占有。但是我们可能会增加执行的时间。

参考文章：http://www.codeproject.com/Articles/27703/Producer-Consumer-Using-Double-Queues

别人在08年就已经想到了，而我却在现在才稍微有点想法。

源码下载

 

后面再大家的评论和建议之下，将代码改为如下：

    public class DoubleQueue {
        private ConcurrentQueue<User> _writeQueue;
        private ConcurrentQueue<User> _readQueue;
        private volatile ConcurrentQueue<User> _currentQueue;

        private AutoResetEvent _dataEvent;

        public DoubleQueue() {
            _writeQueue = new ConcurrentQueue<User>();
            _readQueue = new ConcurrentQueue<User>();
            _currentQueue = _writeQueue;

            _dataEvent = new AutoResetEvent(false);
            Task.Factory.StartNew(() => ConsumerQueue(), TaskCreationOptions.None);
        }

        public void ProducerFunc(User user) {
            _currentQueue.Enqueue(user);
            _dataEvent.Set();
        }

        public void ConsumerQueue() {
            ConcurrentQueue<User> consumerQueue;
            User user;
            int allcount = 0;
            Stopwatch watch = Stopwatch.StartNew();
            while (true)
            {
                _dataEvent.WaitOne();
                if (!_currentQueue.IsEmpty)
                {
                    _currentQueue = (_currentQueue == _writeQueue) ? _readQueue : _writeQueue;
                    consumerQueue = (_currentQueue == _writeQueue) ? _readQueue : _writeQueue;
                    while (!consumerQueue.IsEmpty)
                    {
                        while (!consumerQueue.IsEmpty)
                        {
                            if (consumerQueue.TryDequeue(out user))
                            {
                                FluentConsole.White.Background.Red.Line(user.ToString());
                                allcount++;
                            }
                        }
                        FluentConsole.White.Background.Red.Line($"当前个数{allcount.ToString()}，花费了{watch.ElapsedMilliseconds.ToString()}ms;");
                        System.Threading.Thread.Sleep(20);
                    }
                }
            }
        }
    }