不为空时处理 ConcurrentStack？

Question

我有一个要向其中倾倒项目的 ConcurrentStack。当堆栈不为空时，一次处理这些项目的好方法是什么？我想以一种在未处理堆栈时不会占用 CPU 周期的方式来执行此操作。

我目前得到的基本上是这个，它似乎不是一个理想的解决方案。

private void AddToStack(MyObj obj)
{
    stack.Push(obj);
    HandleStack();
}

private void HandleStack()
{
    if (handling)
        return;

    Task.Run( () =>
    {
        lock (lockObj)
        {
            handling = true;
            if (stack.Any())
            {
                //handle whatever is on top of the stack
            }
            handling = false;
        }
    }
}

所以 bool 存在，因此多个线程不会在等待锁时得到备份。但我不希望同时处理堆栈的多个事物因此锁定。因此，如果两个单独的线程确实最终同时调用 HandleStack 并通过了布尔值，那么锁就在那里，所以两个线程都不会同时遍历堆栈。但是一旦第二个通过锁，堆栈将是空的并且不做任何事情。所以这最终给了我我想要的行为。

所以实际上我只是在 ConcurrentStack 周围编写一个伪并发包装器，而且似乎必须有一种不同的方法来实现这一点。想法？

Answer 1

ConcurrentStack<T> 是实现IProducerConsumerCollection<T> 的集合之一，因此可以被BlockingCollection<T> 包装。 BlockingCollection<T> 有几个方便成员用于常见操作，例如“在堆栈不为空时使用”。例如，您可以循环调用TryTake。或者，您可以使用GetConsumingEnumerable：

private BlockingCollection<MyObj> stack;
private Task consumer;

Constructor()
{
  stack = new BlockingCollection<MyObj>(new ConcurrentStack<MyObj>());
  consumer = Task.Run(() =>
  {
    foreach (var myObj in stack.GetConsumingEnumerable())
    {
      ...
    }
  });
}

private void AddToStack(MyObj obj)
{
  stack.Add(obj);
}

Answer 2

您可以考虑使用Microsoft TPL Dataflow 来做这种事情。

这是一个简单的示例，展示了如何创建队列。试一试，调整MaxDegreeOfParallelism 和BoundedCapacity 的设置，看看会发生什么。

对于您的示例，如果您不希望多个线程同时处理一个数据项，我认为您需要将 MaxDegreeOfParallelism 设置为 1。

（注意：您需要使用 .Net 4.5x 并使用 Nuget 为项目安装 TPL Dataflow。）

还可以阅读Stephen Cleary's blog about TPL。

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace SimpleTPL
{
    class MyObj
    {
        public MyObj(string data)
        {
            Data = data;
        }

        public readonly string Data;
    }

    class Program
    {
        static void Main()
        {
            var queue = new ActionBlock<MyObj>(data => process(data), actionBlockOptions());
            var task = queueData(queue);

            Console.WriteLine("Waiting for task to complete.");
            task.Wait();
            Console.WriteLine("Completed.");
        }

        private static void process(MyObj data)
        {
            Console.WriteLine("Processing data " + data.Data);
            Thread.Sleep(200); // Simulate load.
        }

        private static async Task queueData(ActionBlock<MyObj> executor)
        {
            for (int i = 0; i < 20; ++i)
            {
                Console.WriteLine("Queuing data " + i);
                MyObj data = new MyObj(i.ToString());

                await executor.SendAsync(data);
            }

            Console.WriteLine("Indicating that no more data will be queued.");

            executor.Complete(); // Indicate that no more items will be queued.

            Console.WriteLine("Waiting for queue to empty.");

            await executor.Completion; // Wait for executor queue to empty.
        }

        private static ExecutionDataflowBlockOptions actionBlockOptions()
        {
            return new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 4,
                BoundedCapacity        = 8
            };
        }
    }
}

Answer 3

看起来你想要一个典型的生产者消费者。

我建议使用自动重置事件

当堆栈为空时让您的消费者等待。调用生产者方法时调用Set。

阅读本帖

Fast and Best Producer/consumer queue technique BlockingCollection vs concurrent Queue