c# - 如何迭代 BufferBlock 中的项目？

Question

我最近开始使用 .NET 4.5 中的 TPL 数据流库， block 的整个概念对我来说都是全新的。我在我的应用程序中实现了一个生产者-消费者队列，我需要防止将重复的消息放入队列中，因此需要检查消息是否已经排队。我正在使用 BufferBlock<Message>类型(Message 是自定义类型)。 BufferBlock具有 Count 属性，但这对解决此问题没有帮助，因为需要唯一标识消息。

有什么方法可以检查 BufferBlock 是否存在？包含一个项目或检查所有项目并检查它们？是否可以投 BufferBlock允许迭代项目的东西？我正在关注 example I saw on MSDN它不检查项目是否在队列中，但我认为检查队列的内容是一个相当需要的操作。感谢您的帮助。

Answer 1

而不是闯入 BufferBlock , 为什么不插入一个 TransformManyBlock进入为你做这件事的链条？您可以使用 HashSet , 其中Add方法只返回 true如果尚未添加该项目。它最终变得非常简单，但存储需求显然随着时间的推移而增加......

void Main()
{
    var bb = new BufferBlock<string>();
    var db = DataflowEx.CreateDistinctBlock<string>();
    var ab = new ActionBlock<string>(x => Console.WriteLine(x));
    bb.LinkTo(db);
    db.LinkTo(ab);
    bb.Post("this");
    bb.Post("this");
    bb.Post("this");
    bb.Post("is");
    bb.Post("is");
    bb.Post("a");
    bb.Post("test");
}

public class DataflowEx
{
    public static TransformManyBlock<T, T> CreateDistinctBlock<T>()
    {
        var hs = new HashSet<T>();
        //hs will be captured in the closure of the delegate
        //supplied to the TransformManyBlock below and therefore
        //will have the same lifespan as the returned block.
        //Look up the term "c# closure" for more info
        return new TransformManyBlock<T, T>(
                         x => Enumerable.Repeat(x, hs.Add(x) ? 1 : 0));
    }
}

这样做的原因是，就像 Linq 的 SelectMany 一样，TransformManyBlock 有效地展平了列表的列表。因此，TransformManyBlock 接受一个返回 IEnumerable<T> 的委托(delegate)。 , 但提供返回的项目 IEnumerable<T>一次一个。通过返回 IEnumerable<T>里面有 0 或 1 个项目，我们可以有效地创建 Where -like 行为，允许一个项目通过或阻止它通过，这取决于是否满足某些谓词。在这种情况下，谓词是我们是否可以将项目添加到捕获的 HashSet。