c# - 使用 BlockingCollection 作为单一生产者、单一消费者 FIFO 查询是否好？

Question

我需要单一生产者、单一消费者 FIFO 查询，因为

• 我需要按照收到的顺序处理消息。
• 我需要异步执行此操作，因为调用者不应在我处理消息时等待。
• 只有在上一个消息处理完成后才能开始下一个消息处理。有时“接收”消息的频率高于“处理”消息的频率。但平均而言，我应该能够处理所有消息，只是有时我必须将它们“排队”。

所以我认为它很像 TCP/IP，其中有一个生产者和一个消费者，有时您接收消息的速度比处理速度快，因此您必须查询它们。顺序很重要，而调用者绝对不感兴趣你用那些东西做什么。

这听起来很简单，我可能可以为此使用通用的 Queue，但我想为此使用 BlockingCollection，因为我不想编写任何代码ManualResetEvent 等

BlockingCollection 是否适合我的任务，也许您可​​以提出其他建议？

Answer 1

BlockingCollection 类实现 IProducerConsumerCollection界面非常适合您的要求。

您可以创建两个任务，一个用于异步生产者，另一个作为消费者 worker 。前者会将项目添加到 BlockingCollection，而后者只会在 FIFO 顺序中的新项目可用时立即消耗。

使用 TPL Tasks 的生产者-消费者示例应用程序和 BlockingCollection :

class ProducerConsumer
{
    private static BlockingCollection<string> queue = new BlockingCollection<string>();

    static void Main(string[] args)
    {
        Start();
    }

    public static void Start()
    {
        var producerWorker = Task.Factory.StartNew(() => RunProducer());
        var consumerWorker = Task.Factory.StartNew(() => RunConsumer());

        Task.WaitAll(producerWorker, consumerWorker);
    }

    private static void RunProducer()
    {
        int itemsCount = 100;

        while (itemsCount-- > 0)
        {
            queue.Add(itemsCount + " - " + Guid.NewGuid().ToString());
            Thread.Sleep(250);
        }
    }

    private static void RunConsumer()
    {
        foreach (var item in queue.GetConsumingEnumerable())
        {
           Console.WriteLine(DateTime.Now.ToString("HH:mm:ss.ffff") + " | " + item);
        }
    }
}

IProducerConsumerCollection :

Defines methods to manipulate thread-safe collections intended for producer/consumer usage. This interface provides a unified representation for producer/consumer collections so that higher level abstractions such as System.Collections.Concurrent.BlockingCollection(Of T) can use the collection as the underlying storage mechanism.