BlockingCollection实现单体程序内队列

Posted 2021-02-28 fanfan-90

BlockingCollection可实现单体程序内队列功能，适合写日志、发邮件等并发高可异步的操作

BlockingCollection是一个线程安全的集合类，可提供以下功能：

• 实现制造者-使用者模式。

• 通过多线程并发添加和获取项。

• 可选最大容量。

• 集合为空或已满时通过插入和移除操作进行阻塞。

• 插入和移除“尝试”操作不发生阻塞，或在指定时间段内发生阻塞。

• 使用取消标记执行取消操作。

 

    class Program
    {
        private static readonly BlockingCollection<string> _blockingQueue =
           new BlockingCollection<string>();
        static void Main(string[] args)
        {
            Task.WaitAll(Task.Run(() => { Produce(); }), Task.Run(() => { Consume(); }));


            Console.ReadKey();
        }
        private static void Produce()
        {
            try
            {
                if (!_blockingQueue.IsAddingCompleted)
                {
                    for (int i = 0; i < 100; i++) //限制生产1000次
                    {
                        var now = DateTime.Now.ToString(CultureInfo.InvariantCulture);
                        Console.WriteLine($"第{i + 1}次生产! {now}");
                        _blockingQueue.Add(now);
                    }
                }
            }
            catch
            {
                _blockingQueue.CompleteAdding(); //标记生产完成
            }
            

        }
        private static void Consume()
        {
            int i = 1;
            try
            {
                while (!_blockingQueue.IsCompleted)
                {
                    var x = _blockingQueue.Take();
                    Console.WriteLine($"第{i}次消费 {x}");
                    i++;
                    Thread.Sleep(20);  //故意减慢消费
                }
            }
            catch
            {
                _blockingQueue.CompleteAdding();
            }
            Console.WriteLine("异常，结束！！！");
        }


        private static void DbUpDemo()
        {
            List<string> connectionStringList = new List<string> { "Data Source=192.168.0.116,30705;uid=sa;pwd=KI68oecJc0NpXwscxybK;Initial Catalog=test3;Pooling=true;Max Pool Size=1000;Min Pool Size=5;Connection Timeout=28800" };
            string updateDbMessage = string.Empty;
            Console.WriteLine("更新数据库 开始。。。。");
            bool updateDbResult = UpdateDBHelper.UpdateDb(connectionStringList, out updateDbMessage);
            if (!updateDbResult)
            {
                Console.WriteLine($"更新数据库 失败!-{updateDbMessage}");
            }
            else
            {
                Console.WriteLine("更新数据库 成功!");
            }

        }
    }

 

案例1：使用BlockingCollection做日志记录：

 public class DBLoggerProcessor : IDisposable
    {
        private readonly BlockingCollection<string> _messageQueue = new BlockingCollection<string>(1024);

        private readonly Thread _writerThread;

        public DBLoggerProcessor()
        {
            // Start DB Logger queue processor
            _writerThread = new Thread(ProcessLogQueue)
            {
                IsBackground = true,
                Name = "DB logger queue processing thread"
            };
            _writerThread.Start();
        }

        public virtual void EnqueueMessage(string message)
        {
            if (!_messageQueue.IsAddingCompleted)
            {
                try
                {
                    _messageQueue.Add(message);
                    return;
                }
                catch (InvalidOperationException) { }
            }
            WriteMessage(message);
        }

        internal virtual void WriteMessage(string message)
        {
           //记录数据库
        }
        private void ProcessLogQueue()
        {
            try
            {
                foreach (var message in _messageQueue.GetConsumingEnumerable())
                {
                    WriteMessage(message);
                }
            }
            catch
            {
                try
                {
                    _messageQueue.CompleteAdding();
                }
                catch { }
            }
        }
        public void Dispose()
        {
            _messageQueue.CompleteAdding();

            try
            {
                _writerThread.Join(1500);
            }
            catch (ThreadStateException) { }
        }
    }