对于一个快速生产者多个慢速消费者，最佳方案是什么?

本文关键字：最佳 方案 消费者 是什么 于一个 生产者 | 更新日期: 2023-09-27 18:33:29

我正在寻找实现一个生产者多使用者多线程应用程序的最佳方案。目前，我正在为共享缓冲区使用一个队列，但它比一个生产者一个消费者的情况要慢得多。我打算这样做：

Queue<item>[] buffs = new Queue<item>[N];
object[] _locks = new object[N];
static void Produce()
{
    int curIndex = 0;
    while(true)
    {
        // Produce item;
        lock(_locks[curIndex])
        {
            buffs[curIndex].Enqueue(curItem);
            Monitor.Pulse(_locks[curIndex]);
        }
        curIndex = (curIndex+1)%N;
    }
}
static void Consume(int myIndex)
{
    item curItem;
    while(true)
    {
        lock(_locks[myIndex])
        {
            while(buffs[myIndex].Count == 0)
                Monitor.Wait(_locks[myIndex]);
            curItem = buffs[myIndex].Dequeue();
        }
        // Consume item;
    }
}
static void main()
{
    int N = 100;
    Thread[] consumers = new Thread[N];
    for(int i = 0; i < N; i++)
    {
        consumers[i] = new Thread(Consume);
        consumers[i].Start(i);
    }
    Thread producer = new Thread(Produce);
    producer.Start();
}

对于一个快速生产者多个慢速消费者，最佳方案是什么?

使用阻塞集合

BlockingCollection<item> _buffer = new BlockingCollection<item>();
static void Produce()
{
    while(true)
    {
        // Produce item;
        _buffer.Add(curItem);
    }
    // eventually stop producing
    _buffer.CompleteAdding();
}
static void Consume(int myIndex)
{
    foreach (var curItem in _buffer.GetConsumingEnumerable())
    {
        // Consume item;
    }
}
static void main()
{
    int N = 100;
    Thread[] consumers = new Thread[N];
    for(int i = 0; i < N; i++)
    {
        consumers[i] = new Thread(Consume);
        consumers[i].Start(i);
    }
    Thread producer = new Thread(Produce);
    producer.Start();
}

如果不想从一开始就指定线程数，可以改用 Parallel.ForEach。

static void Consume(item curItem)
{
    // consume item
}
void Main()
{
    Thread producer = new Thread(Produce);
    producer.Start();
    Parallel.ForEach(_buffer.GetConsumingPartitioner(), Consumer)
}

使用更多线程无济于事。它甚至可能降低性能。我建议您尝试使用每个工作项都是由生产者创建的一个项的ThreadPool。但是，这并不能保证生产的物品按其生产的顺序消耗。

另一种方法是将消费者的数量减少到 4 个，例如，并修改它们的工作方式，如下所示：

制作人将新作品添加到队列中。所有工作线程只有一个全局队列。然后，它设置一个标志来指示有这样的新工作：

ManualResetEvent workPresent = new ManualResetEvent(false);
Queue<item> workQueue = new Queue<item>();
static void Produce()
{
    while(true)
    {
        // Produce item;
        lock(workQueue)
        {
            workQueue.Enqueue(newItem);
            workPresent.Set();
        }
    }
}

使用者等待将工作添加到队列中。只有一个消费者可以完成它的工作。然后，它从队列中获取所有工作并重置标志。在此之前，制片人将无法添加新作品。

static void Consume()
{
    while(true)
    {
        if (WaitHandle.WaitOne(workPresent))
        {
            workPresent.Reset();
            Queue<item> localWorkQueue = new Queue<item>();
            lock(workQueue)
            {
                while (workQueue.Count > 0)
                    localWorkQueue.Enqueue(workQueue.Dequeue());
            }
            // Handle items in local work queue
            ...
        }
    }
}

然而，这种结果有点不可预测。可能是某个线程正在执行所有工作，而其他线程什么都不做。

我不明白为什么你必须使用多个队列。只需减少锁定量即可。下面是一个示例，您可以在其中拥有大量消费者，他们都在等待新的工作。

public class MyWorkGenerator
{
    ConcurrentQueue<object> _queuedItems = new ConcurrentQueue<object>();
    private object _lock = new object();
    public void Produce()
    {
        while (true)
        {
            _queuedItems.Enqueue(new object());
            Monitor.Pulse(_lock);
        }
    }
    public object Consume(TimeSpan maxWaitTime)
    {
        if (!Monitor.Wait(_lock, maxWaitTime))
            return null;
        object workItem;
        if (_queuedItems.TryDequeue(out workItem))
        {
            return workItem;
        }
        return null;
    }
}

请注意，Pulse()一次只会触发一个使用者。

用法示例：

    static void main()
    {
        var generator = new MyWorkGenerator();
        var consumers = new Thread[20];
        for (int i = 0; i < consumers.Length; i++)
        {
            consumers[i] = new Thread(DoWork);
            consumers[i].Start(generator);
        }
        generator.Produce();
    }
    public static void DoWork(object state)
    {
        var generator = (MyWorkGenerator) state;
        var workItem = generator.Consume(TimeSpan.FromHours(1));
        while (workItem != null)
        {
            // do work

            workItem = generator.Consume(TimeSpan.FromHours(1));
        }
    }

请注意，实际队列隐藏在生产者中，因为它是实现细节。使用者实际上不必知道工作项是如何生成的。

相关文章：