异步简析之BlockingCollection实现生产消费模式

Posted linkanyway

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了异步简析之BlockingCollection实现生产消费模式相关的知识,希望对你有一定的参考价值。

目前市面上有诸多的产品实现队列功能,比如Redis、MemCache等...
其实c#中也有一个基础的集合类专门用来实现生产/消费模式 (生产模式还是建议使用Redis等产品)

下面是官方的一些资料和介绍:

BlockingCollection 是一个线程安全集合类,可提供以下功能:

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

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

  • 可选最大容量。

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

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

  • 封装实现 IProducerConsumerCollection 的任何集合类型

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

  • 支持使用 foreach(在 Visual Basic 中,使用 For Each)的两种枚举:

    • 只读枚举。

    • 在枚举项时将项移除的枚举。

BlockingCollection 支持限制和阻塞。 限制意味着可以设置集合的最大容量。 限制在某些情况中很重要,因为它使你能够控制内存中的集合的最大大小,并可阻止制造线程移动到离使用线程前方太远的位置。

可以看到该类是完全线程安全的,因此用来做生产/消费是非常合适的

如下代码演示了该类在异步环境中很好的执行着生产和消费任务

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
using Xunit;

namespace AsyncLearning
{
    public class TestProduceAndConsumer
    {
        /// <summary>
        /// BlockingCollection是线程安全的集合类型,支持多线程同时读写
        /// </summary>
        private readonly BlockingCollection<string> _blockingQueue =
            new BlockingCollection<string>();


        private void Produce()
        {
            for (int i = 0; i < 10; i++) //限制生产1000次
            {
                var now = DateTime.Now.ToString(CultureInfo.InvariantCulture);
                Debug.WriteLine($"第{i+1}次生产! {now}");
                _blockingQueue.Add(now);
                Thread.Sleep(1000);  //特意减慢生产过程以至于不会太快。。。方便演示
            }


            _blockingQueue.CompleteAdding(); //标记生产完成
        }


        private void Consume()
        {
            int i = 1;
            while (!_blockingQueue.IsCompleted)
            {
                var x = _blockingQueue.Take();
                Debug.WriteLine($"第{i}次消费 {x}");
                i++;
                Thread.Sleep(2000);  //故意减慢消费
            }
        }


        [Fact]
        public void Test()
        {
            
            Task.WaitAll(Task.Run(() => { Produce(); }), Task.Run(() => { Consume(); }));
        }
    }
}

输出如下:
技术分享图片?

可见Demo很好的按照设定执行了代码逻辑,由于故意设定了不同的sleep时间,可以看到消费是晚于生产的,而消费全部完成后本Demo的任务全部结束

 



以上是关于异步简析之BlockingCollection实现生产消费模式的主要内容,如果未能解决你的问题,请参考以下文章

创建异步任务队列 - 使用 BlockingCollection C#

[tomcat]源码简析 异步/非阻塞和请求构成

[tomcat]源码简析 异步/非阻塞和请求构成

如何加快大块 BlockingCollection 的实现

Flutter之SchedulerBinding简析

Vue源码之 virtual-dom 实现简析