BlockingCollection线程安全吗？

Question

BlockingCollection<int> blockingCollection = new BlockingCollection<int>();
    // create and start a producer
    Task.Factory.StartNew(() => {
        // put items into the collectioon
        for (int i = 0; i < 1000; i++)
        {
            blockingCollection.Add(i);
        }
        // mark the collection as complete
        blockingCollection.CompleteAdding();
    });
    // create and start a producer
    Task.Factory.StartNew(() => {
        while (!blockingCollection.IsCompleted)
        {
            // take an item from the collection
            int item = blockingCollection.Take();
            // print out the item
            Console.WriteLine("Taskid{1} Item {0}", item, Task.CurrentId);
        }
    });

    // create and start a producer
    Task.Factory.StartNew(() => {
        while (!blockingCollection.IsCompleted)   // if the blockingCollection is not completed
        {
            // take an item from the collection
            int item = blockingCollection.Take();  // but in here, all the items have been taken by other thread, this line will wait forever?
            // print out the item
            Console.WriteLine("Taskid{1} Item {0}", item,Task.CurrentId);
        }
    });
    Console.ReadLine();

关键代码行：

    while (!blockingCollection.IsCompleted)   // if the blockingCollection is not completed
    int item = blockingCollection.Take();  // but in here, all the items have been taken by other thread, this line will wait forever?

Answer 1

问题1-线程安全吗？

引用BlockingCollection文档

为实现线程安全的集合提供阻塞和绑定功能。

问题2-- Take()会永远用吗？

来自正式文档：MS关于阻塞收集-获取

OperationCanceledException BlockingCollection是空的，并已被标记为完整的添加。

因此，当在已完成的Take上调用BlockingCollection时，将引发异常，您可以相应地处理该异常。

Answer 2

没错，BlockingCollection<T>是线程安全的。

为了避免IsCompleted检查，可以在GetConsumingEnumerable()的帮助下放置相同的逻辑，直到阻塞集合完成为止：

  // Producer:

  Task.Run(() => { 
    for (int i = 0; i < 1000; i++)
      blockingCollection.Add(i);

    blockingCollection.CompleteAdding();
  });    

  // Consumers:

  Task.Run(() => {
    foreach (var item in blockingCollection.GetConsumingEnumerable()) 
      Console.WriteLine("Taskid{1} Item {0}", item, Task.CurrentId);
  });

  Task.Run(() => {
    foreach (var item in blockingCollection.GetConsumingEnumerable()) 
      Console.WriteLine("Taskid{1} Item {0}", item, Task.CurrentId);
  });

Answer 3

BlockingCollection类是线程安全的，因为当多个线程并发调用时，它的内部状态是不受损坏的。从某种意义上说，它的存在并不是线程安全的，因为它的存在与线程安全有关--线程不安全的代码块！

您的代码是不安全的，因为在调用IsCompleted和Take之间，另一个线程可以调用Complete方法，并将BlockingCollection的状态从未完成更改为已完成。线程安全解决方案是在原子操作中使用将IsCompleted和Take结合在一起的API。BlockingCollection类中没有这样的API，但是您可以使用GetConsumingEnumerable方法实现您想要的结果。此方法返回一个在等待可用项时阻塞的IEnumerable，并在BlockingCollection完成时完成。实际上，这是使用此集合的标准和首选方法。