c# - 丢弃旧数据的 BlockingCollection

Question

我有一个 BlockingCollection .生产者任务向其中添加项目，消费者任务移除项目。

现在我想限制集合中的项目数量，如果添加了更多项目，则自动丢弃旧数据。该集合不应同时包含超过 N 个最近添加的项目。

因此，如果生产者添加新项目的速度比消费者移除新项目的速度快，我希望消费者只处理最新的项目。

我可以在其构造函数中限制 BlockingCollection 的大小，但这当然意味着它会在添加更多项目时阻塞，而不是删除旧项目。

(我不想在生产者端阻塞，只有消费者端在从空集合中检索项目时应该阻塞。)

我目前的解决方案是 hack，并且只适用于 1 的大小限制:
(而且我不太确定它是否可靠。)

// My consumer task:
foreach (var item in blockingCollection.GetConsumingEnumerable())
{
    var lastItem = item;
    var lastItemTmp = item;
    while (blockingCollection.TryTake(out lastItemTmp))
           lastItem = lastItemTmp;
    // Now lastItem contains the most recent item in the collection, 
    // and older items have been discarded.
    // Proceed consuming lastItem ...
}

是否有更清洁的解决方案？

Answer 1

这样做:

void AddItemToQueue(MyClass item)
{
    while (!queue.TryAdd(item))
    {
        MyClass trash;
        queue.TryTake(out trash);
    }
}

如果在您尝试添加项目时队列已满，则会从队列中删除一个项目。它使用 TryTake 因为有可能(不太可能，但有可能)其他线程可能在您有机会取一个之前从队列中删除了最后一个项目。

当然，这假设您在构造 BlockingCollection 时指定了项目数量限制。

实现此目的的另一种方法是创建您自己的循环队列类，并让它实现 IProducerConsumerCollection界面。然后，您可以使用该类的实例作为 BlockingCollection 的后备集合。实现循环队列并不是特别困难，尽管有些边缘情况很难正确处理。而且您必须使它成为并发数据结构，尽管使用锁很容易做到这一点。

如果您不希望队列经常溢出，或者如果队列的流量非常低(即每秒不会被命中数千次)，那么我最初的建议将按照您的意愿行事并且不会性能问题。如果存在性能问题，那么循环队列就是解决方案。