将 BlockingCollection<T> 用作单生产者、单消费者 FIFO 查询是否很好？

Question

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

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

所以我认为这很像 TCP/IP，你有一个生产者和一个消费者，有时你接收消息的速度比你处理的速度要快，所以你必须查询它们。订单在哪里很重要，而呼叫者对你用这些东西做什么完全不感兴趣。

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

BlockingCollection 对我的任务有多合适？或许您可以提出其他建议？

Answer 1

BlockingCollection 类实现了IProducerConsumerCollection 接口，非常适合您的要求。

您可以创建两个任务，一个用于异步生产者，另一个作为消费者工作者。前者会将项目添加到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:

定义用于操作线程安全集合的方法 生产者/消费者使用。该接口提供了一个统一的 生产者/消费者集合的表示，以便更高级别 抽象如 System.Collections.Concurrent.BlockingCollection(Of T) 可以使用 集合作为底层存储机制。

Answer 2

既然是你需要的队列，为什么不坚持一个队列呢？您可以使用Syncrhonized Queue。