等到后台线程清除BlockingCollection队列,如果花费的时间太长,会超时吗？

Question

在C#中,我想知道是否可以等到后台线程清除BlockingCollection,如果时间太长则超时.

我现在拥有的临时代码让我觉得有些不雅(因为什么时候使用它是好的做法Thread.Sleep？):

while (_blockingCollection.Count > 0 || !_blockingCollection.IsAddingCompleted)
{
    Thread.Sleep(TimeSpan.FromMilliseconds(20));
    // [extra code to break if it takes too long]
}

Answer 1

您可以使用GetConsumingEnumerable()和foreach在消费线程中确定队列何时为空,然后设置ManualResetEvent主线程可以检查哪个队列为空.GetConsumingEnumerable()返回一个枚举器,它检查CompleteAdding()在空队列终止之前是否已被调用.

示例代码:

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    internal class Program
    {
        private void run()
        {
            Task.Run(new Action(producer));
            Task.Run(new Action(consumer));

            while (!_empty.WaitOne(1000))
                Console.WriteLine("Waiting for queue to empty");

            Console.WriteLine("Queue emptied.");
        }

        private void producer()
        {
            for (int i = 0; i < 20; ++i)
            {
                _queue.Add(i);
                Console.WriteLine("Produced " + i);
                Thread.Sleep(100);
            }

            _queue.CompleteAdding();
        }

        private void consumer()
        {
            foreach (int n in _queue.GetConsumingEnumerable())
            {
                Console.WriteLine("Consumed " + n);
                Thread.Sleep(200);
            }

            _empty.Set();
        }

        private static void Main()
        {
            new Program().run();
        }

        private BlockingCollection<int> _queue = new BlockingCollection<int>();

        private ManualResetEvent _empty = new ManualResetEvent(false);
    }
}

Answer 2

如果你在你的消费线程中写这样的东西怎么办：

var timeout = TimeSpan.FromMilliseconds(10000);
T item;
while (_blockingCollection.TryTake(out item, timeout))
{
     // do something with item
}
// If we end here. Either we have a timeout or we are out of items.
if (!_blockingCollection.IsAddingCompleted)
   throw MyTimeoutException();