前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >.NET(C#):线程安全集合的阻塞BlockingCollection的使用

.NET(C#):线程安全集合的阻塞BlockingCollection的使用

作者头像
zls365
发布2021-01-13 15:53:02
1.4K0
发布2021-01-13 15:53:02
举报
文章被收录于专栏:CSharp编程大全CSharp编程大全

目录

  • 1. 限制最大容量:BoundedCapacity
  • 2. 禁止加入:CompleteAdding和IsCompleted
  • 3. 枚举:GetConsumingEnumerable和BlockingCollection本身
  • 4. GetConsumingEnumerable和CompleteAdding

返回目录

1. 限制最大容量:BoundedCapacity

BoundedCapacity属性和CompleteAdding方法,它们都可以从某种方式上限制元素被加入到集合中。但BoundedCapacity是用来限制集合的最大容量,当容量已满后,后续的添加操作会被阻塞,一旦有元素被移除,那么阻塞的添加操作会成功执行。

比如下面代码,试图将1-50加入到BlockingCollection,此时默认内部是ConcurrentBag,当然你可以指定任意IProducerConsumerCollection。我们把BoundedCapacity设置成2。

代码语言:javascript
复制
var bcollec =newBlockingCollection<int>(2);
//试图添加1-50
Task.Run(() =>
    {
        Parallel.For(1, 51, i =>
        {
            bcollec.Add(i);
            Console.WriteLine("加入:"+ i);
        });
    });
 
Thread.Sleep(1000);
Console.WriteLine("调用一次Take");
bcollec.Take();
 
Thread.Sleep(Timeout.Infinite);

可能的输出:

加入:37

加入:13

调用一次Take

加入:25

只有最多两个可以加入,然后调用Take后,下一个元素才会被加入。(注意此时Parallel.For中会有多个线程处于阻塞状态,因为无法加入数据)。

返回目录

2. 禁止加入:CompleteAdding和IsCompleted

CompleteAdding方法则是直接不允许任何元素被加入集合,即使是当前元素的数量小于BoundedCapacity属性。

代码:

代码语言:javascript
复制
var bcollec =newBlockingCollection<int>(5);
//试图添加1-50
Task.Run(() =>
    {
        Parallel.For(1, 51, i =>
        {
            Console.WriteLine("准备加入:"+ i);
            bcollec.Add(i);
            Console.WriteLine("== 成功加入:"+ i);
            Thread.Sleep(1000);
        });
    });
 
//等待一小段时间后马上调用CompleteAdding
Thread.Sleep(500);
Console.WriteLine("调用CompleteAdding");
bcollec.CompleteAdding();
 
Thread.Sleep(Timeout.Infinite);

上述代码可能的输出:

准备加入:1

准备加入:13

准备加入:25

准备加入:37

== 成功加入:13

== 成功加入:1

== 成功加入:37

== 成功加入:25

调用CompleteAdding

准备加入:2

准备加入:38

准备加入:26

准备加入:14

可以看到,虽然BlockingCollection的BoundedCapacity为5,但是由于提前调用了CompleteAdding,即使当前集合只有4个元素,也不会再同意新的加入操作了。

那么CompleteAdding有什么用?当使用了CompleteAdding方法后且集合内没有元素的时候,另一个属性IsCompleted此时会为True,这个属性可以用来判断是否当前集合内的所有元素都被处理完,而BlockingCollection背后的IProducerConsumerCollection恰恰常用来处理此类生产者-消费者问题的。

下面我们首先在多个线程中试图往BlockingCollection中加入元素,然后中途调用CompleteAdding,接着通过IsCompleted属性逐个处理被成功加入的元素。

如下代码:

代码语言:javascript
复制
var bcollec =newBlockingCollection<int>();
//试图添加1-50
Task.Run(() =>
    {
        Parallel.For(1, 51, i =>
        {
            bcollec.Add(i);
            Console.WriteLine("成功加入:"+ i);
            Thread.Sleep(1000);
        });
    });
 
//等待一小段时间后马上调用CompleteAdding
Thread.Sleep(700);
Console.WriteLine("调用CompleteAdding");
bcollec.CompleteAdding();
 
Console.WriteLine("容器元素数量:"+ bcollec.Count);
 
while (!bcollec.IsCompleted)
{
    var res = bcollec.Take();
    Console.WriteLine("取出:"+ res);
}
 
Console.WriteLine("操作完成");
 
Thread.Sleep(Timeout.Infinite);

可能的输出:

成功加入:37

成功加入:25

成功加入:13

成功加入:1

调用CompleteAdding

容器元素数量:4

取出:1

取出:37

取出:25

取出:13

操作完成

返回目录

3. 枚举:GetConsumingEnumerable和BlockingCollection本身

BlockingCollection有两种枚举方法,首先BlockingCollection本身继承自IEnumerable<T>,所以它自己就可以被foreach枚举,首先BlockingCollection包装了一个线程安全集合,那么它自己也是线程安全的,而当多个线程在同时修改或访问线程安全容器时,BlockingCollection自己作为IEnumerable会返回一个一定时间内的集合片段,也就是只会枚举在那个时间点上内部集合的元素。

看下面代码:

代码语言:javascript
复制
var bcollec =newBlockingCollection<int>();
//试图添加1-10
Task.Run(() =>
{
    var forOpt =newParallelOptions()
    {
        //防止在某些硬件上并发数太多
        MaxDegreeOfParallelism =2
    };
 
    Parallel.For(1, 11, forOpt, i =>
    {
        bcollec.Add(i);
        Console.WriteLine("成功加入:"+ i);
        Thread.Sleep(500);
    });
});
 
Thread.Sleep(700);
//开始枚举
Task.Run(() =>
{
    foreach (var i in bcollec)
        Console.WriteLine("输出:"+ i);
});
 
Thread.Sleep(Timeout.Infinite);

我们边加入元素边进行枚举(直接在BlockingCollection上foreach),可能的输出:

成功加入:1

成功加入:6

成功加入:2

成功加入:7

输出:1

输出:6

输出:2

输出:7

成功加入:8

成功加入:3

成功加入:4

成功加入:9

成功加入:5

成功加入:10

可以看到,BlockingCollection本身的迭代器只能反映出一时的容器内容。

而BlockingCollection还有一个GetConsumingEnumerable方法,同样返回一个IEnumerable<T>,这个可枚举的集合背后的迭代器不同于BlockingCollection本身的迭代器,它可以返回最新的加入的元素,如果当前时间段没有元素被加入,它会阻塞然后等新加进来的元素。

我们把上面的使用BlockingCollection本身枚举代码中的枚举Task改成这样:

代码语言:javascript
复制
//开始枚举
Task.Run(() =>
{
    foreach (var i in bcollec.GetConsumingEnumerable())
        Console.WriteLine("输出:"+ i);
    Console.WriteLine("完成枚举");
});

可能的输出:

成功加入:6

成功加入:1

成功加入:2

成功加入:7

输出:6

输出:1

输出:2

输出:7

成功加入:3

成功加入:8

输出:3

输出:8

成功加入:4

成功加入:9

输出:4

输出:9

成功加入:10

成功加入:5

输出:10

输出:5

这个迭代器很给力,一直处于等待和执行的状态,只要有新的元素被加入,它会找机会去执行foreach的内容,然后再阻塞去等新的元素。

而且在输出中,代码里的“完成枚举”字符串一直没有被输出。此时它还在卖力地等……因为它不确定什么时候才不会有新元素被加入。

返回目录

4. GetConsumingEnumerable和CompleteAdding

好,此时你应该想到了上面学的CompleteAdding方法,它可以禁止新的元素被加入到BlockingCollection的内部线程安全集合中,所以使用这个方法可以通知GetConsumingEnumerable的迭代器您老不用再等了,后面不会有元素被加进来了。

如下代码:

抱歉,这几段代码都不短,而且都类似。但我仍然把完整代码贴出来,虽然这使文章比较冗长,但是我觉得这样读者浏览或者复制时从上到下一目了然,总比看到诸如“请把前面xxx个代码做如下修改:把xxx行改成xxx,在xxx行加入这段代码……”好吧。

代码语言:javascript
复制
var bcollec =newBlockingCollection<int>();
//试图添加1-10
Task.Run(() =>
{
    var forOpt =newParallelOptions()
    {
        //防止在某些硬件上并发数太多
        MaxDegreeOfParallelism =2
    };
 
    Parallel.For(1, 11, forOpt, i =>
    {
        Console.WriteLine("等待加入:"+ i);
        bcollec.Add(i);
        Console.WriteLine("成功加入:"+ i);
        Thread.Sleep(500);
    });
});
 
Thread.Sleep(600);
//开始枚举
Task.Run(() =>
{
    foreach (var i in bcollec.GetConsumingEnumerable())
        Console.WriteLine("输出:"+ i);
    Console.WriteLine("完成枚举");
});
 
Thread.Sleep(300);
 
bcollec.CompleteAdding();
Console.WriteLine("=== 调用CompleteAdding");
 
Thread.Sleep(Timeout.Infinite);
 

可能的输出:

等待加入:1

等待加入:6

成功加入:1

成功加入:6

等待加入:2

成功加入:2

等待加入:7

成功加入:7

输出:1

输出:6

输出:2

输出:7

=== 调用CompleteAdding

完成枚举

等待加入:3

等待加入:8

可以看到,等CompleteAdding,“枚举完成”马上被输出!

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2020-12-31,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 CSharp编程大全 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 1. 限制最大容量:BoundedCapacity
  • 2. 禁止加入:CompleteAdding和IsCompleted
  • 3. 枚举:GetConsumingEnumerable和BlockingCollection本身
  • 4. GetConsumingEnumerable和CompleteAdding
相关产品与服务
容器服务
腾讯云容器服务(Tencent Kubernetes Engine, TKE)基于原生 kubernetes 提供以容器为核心的、高度可扩展的高性能容器管理服务,覆盖 Serverless、边缘计算、分布式云等多种业务部署场景,业内首创单个集群兼容多种计算节点的容器资源管理模式。同时产品作为云原生 Finops 领先布道者,主导开源项目Crane,全面助力客户实现资源优化、成本控制。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档