c# - 等到后台线程清除 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()andforeach来判断队列何时为空，然后设置一个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();

Answer 3

如果您可以重新设计以允许在集合为空时设置事件，则可以使用等待句柄来等待集合为空：

static void Main(string[] args)
{
    BlockingCollection<int> bc = new BlockingCollection<int>();
    ManualResetEvent emptyEvent = new ManualResetEvent(false);
    Thread newThread = new Thread(() => BackgroundProcessing(bc, emptyEvent));
    newThread.Start();
    //wait for the collection to be empty or the timeout to be reached.
    emptyEvent.WaitOne(1000);

}
static void BackgroundProcessing(BlockingCollection<int> collection, ManualResetEvent emptyEvent)
{
    while (collection.Count > 0 || !collection.IsAddingCompleted)
    {
        int value = collection.Take();
        Thread.Sleep(100);
    }
    emptyEvent.Set();
}