C# BlockingCollection 生产者消费者不阻塞消费者线程

Question

我有一种情况，我需要有大量（数百个）队列，其中项目应该按顺序处理（需要单线程消费者）。我的第一个实现，based on the samples, 我为每个 BlockingCollection 使用了一个长时间运行的任务来使用队列项。然而，我最终得到了一个拥有数百个线程的应用程序，这些线程大多处于空闲状态，除了消耗内存之外什么都不做，因为队列大部分时间都是空的。

我认为只有在队列中有要处理的内容时才运行消费者任务会更好，但是，我无法找到提供最佳实践的示例。

我想出了一个类似于下面的解决方案。但问题是，每个项目都会导致一个新任务（也许这效率低下？浪费资源？）。但是，如果我不为每个项目创建一个新任务，我不能保证一个项目不会处于未处理的队列中。

    private object _processSyncObj = new object();
    private volatile bool _isProcessing;
    private BlockingCollection<string> _queue = new BlockingCollection<string>();

    private void EnqueueItem(string item)
    {
        _queue.Add(item);
        Task.Factory.StartNew(ProcessQueue);
    }

    private void ProcessQueue()
    {
        if (_isProcessing)
            return;

        lock (_processSyncObj)
        {
             string item;
             while (_isProcessing = _queue.TryTake(out item))
             {
                 // process item
             }
        }
    }

什么是这种情况的最佳实践/最佳解决方案，以保证不存在项目在队列中但没有消费者正在运行的情况？

Answer 1

我认为您所做的事情是合理的，因为 Task 可以很好地扩展数百万个任务，生成针对 ThreadPool 的内部子队列，避免过多的上下文切换。

在幕后，任务排队到线程池，该线程池已通过确定和调整线程数量的算法得到增强，并提供负载平衡以最大限度地提高吞吐量。这使得任务相对轻量级，您可以创建许多任务以实现细粒度并行。

Task Parallelism (Task Parallel Library)

...但是您所做的最终将只是一个普通的任务编程，因为对于每个入队，您都会启动一个任务，因此阻塞集合是完全未使用的。据了解，您关心的是触发任务并让 TaskScheduler 在作业到达时按顺序运行它们。

你知道你也可以自定义TaskScheduler吗？

如果只使用一个任务编程模式，加上一个自定义的TaskScheduler来控制计划任务的流程呢？

例如，您可以创建一个 OrderedTaskScheduler，它派生自一个 LimitedConcurrencyLevelTask​​Scheduler，其行为类似于...

LimitedConcurrencyLevelTask​​Scheduler 类提供了一个任务调度程序，可确保在 ThreadPool 之上运行时达到最大并发级别。需要设置此调度程序所需的最大并行度。

OrderedTaskScheduler 类提供了一个任务调度程序，可确保一次只执行一个任务。任务按照它们排队的顺序执行 (FIFO)。它是 LimitedConcurrencyLevelTask​​Scheduler 的子类，发送 1 作为其基类构造函数的参数。

你可以找到这些已经开发好的调度器，它们被称为ParallelExtensionsExtras，你可以从here下载它，并从blog post和others阅读一些关于它的内容。

您也可以直接在nuget 上找到它，在github 上找到代码镜像。

享受吧！ :)

Answer 2

你考虑过Parallel Extension Extras吗？我相信QueuedTaskScheduler or ThreadPerTaskScheduler 可以轻松满足您的方案。

Answer 3

当然是重写，但你是否考虑过这样做？

public class WorkerQueue<T>
{
    public WorkerQueue(Action<T> workerMethod)
    {
        _workerMethod = workerMethod;
        Task.Factory.StartNew(WorkerAction);
    }

    private Action<T> _workerMethod;

    private void WorkerAction()
    {
        lock (_processSyncObj)
        {
            if (_workerMethod == null)
                return;

            while (true)
            {
                T item;
                if (_queue.TryTake(out item))
                {
                    var method = _workerMethod;
                    if (method != null)
                        method(item);

                }
            }
        }
    }

    private BlockingCollection<T> _queue = new BlockingCollection<T>();
    private object _processSyncObj = new object();
    private volatile bool _isProcessing;

    public void EnqueueItem(T item)
    {
        // thought you might want to swap BlockingCollection with a normal collection since you apparently only want your read threadlocked? You're already making that sure in "WorkerAction"
        _queue.Add(item);
    }
}


/// <summary>
/// Usage example
/// </summary>
public class Program
{
    public void Start()
    {
        var test = new WorkerQueue<string>(WorkerMethod);
    }

    private void WorkerMethod(string s)
    {
        Console.WriteLine(s);
    }
}