如何在管道中阻止任务直到集合已满，然后再从中删除项目答案

【问题标题】：How to make a task in pipeline block till the collection is full before removing items from it如何在管道中阻止任务直到集合已满，然后再从中删除项目
【发布时间】：2015-08-15 13:44:39
【问题描述】：

我有 3 个任务构成了一个管道。我希望终结器任务（管道中的最后一个）仅在达到其边界容量后才开始使用来自 buffer2 的数据。现在，它会在每件物品到达时挑选它们，我不希望这样。

 static void Main(string[] args)
    {
        List<int> input = Enumerable.Range(0, 20).ToList();

        BlockingCollection<int> buffer1 = new BlockingCollection<int>(10);
        BlockingCollection<int> buffer2 = new BlockingCollection<int>(5);

        Task producer = Task.Factory.StartNew(() => Producer(input, buffer1));
        Task consumer = Task.Factory.StartNew(() => Consumer(buffer1, buffer2));
        Task finalizer = Task.Factory.StartNew(() => Finalizer(buffer2));

        Task.WaitAll(producer,consumer, finalizer);
        Console.ReadLine();
    }

    private static void Producer(List<int> input, BlockingCollection<int> buffer1)
    {
        foreach (int i in input)
        {
            buffer1.Add(i);
        }

        buffer1.CompleteAdding();
    }

    private static void Consumer(BlockingCollection<int> buffer1, BlockingCollection<int> buffer2)
    {
        foreach(int i in buffer1.GetConsumingEnumerable())
        {
            Console.WriteLine("Consumer saw item " + i);

            buffer2.Add(i);
        }

        buffer2.CompleteAdding();
    }

    private static void Finalizer(BlockingCollection<int> buffer)
    {
        foreach (int i in buffer.GetConsumingEnumerable())
        {
            // Do some work
            Console.WriteLine("Finalizer saw item " + i);
            System.Threading.Thread.Sleep(1000);
        }

        buffer.CompleteAdding();
    }

如何让终结器在 buffer2 中包含 5 个项目之前停止使用这些项目？消费者也一样，只有在缓冲区 1 至少有 10 个项目时才应该开始消费。

【问题讨论】：

只是好奇。您是否有合法的用例表明此行为有用？
是的。我在磁盘上有一个巨大的 csv 文件。我想读取一大段行，然后在消费者将这些行发送到终结器之前处理这些行，终结器将这些行发送到服务器

标签： c# async-await task-parallel-library task

【解决方案1】：

也许你应该看看TPL Dataflow from Microsoft

我已对您的代码进行了一些更改以适应 TPL

static void Main(string[] args)
{
    List<int> input = Enumerable.Range(0, 20).ToList();

    BatchBlock<int> buffer1 = new BatchBlock<int>(10);
    BatchBlock<int> buffer2 = new BatchBlock<int>(5);
    ActionBlock<int[]> action1;
    ActionBlock<int[]> action2;

    action1 = new ActionBlock<int[]>(t => { Consumer(t, buffer2); },
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });

    buffer1.LinkTo(action1, new DataflowLinkOptions { PropagateCompletion = true });

    action2 = new ActionBlock<int[]>(t => Finalizer(t),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1 });

    buffer2.LinkTo(action2, new DataflowLinkOptions { PropagateCompletion = true });

    Task produceTask = Task.Factory.StartNew(() => Producer(input, buffer1));

    Task.WaitAll(produceTask);
    action1.Completion.Wait();//Will add all the items to buffer2
    buffer2.Complete();//will not get any new items
    action2.Completion.Wait();//Process the batch of 5 and then complete
    Console.WriteLine("Process complete");
    Console.ReadLine();
}

private static void Finalizer(int[] t)
{
    Console.WriteLine("Received a batch of items {0}", t.Count());
    foreach (int i in t)
    {
        // Do some work
        Console.WriteLine("Finalizer saw item " + i);
        System.Threading.Thread.Sleep(1000);
    }
}

private static void Consumer(int[] t, BatchBlock<int> buffer2)
{
    foreach (var item in t)
    {
        Console.WriteLine("Consumer saw item " + item);
        buffer2.Post(item);
    }

}

public static void Producer(List<int> input, BatchBlock<int> buffer1)
{
    foreach (int i in input)
    {
        buffer1.Post(i);
    }

    //Once marked complete your entire data flow will signal a stop for
    // all new items
    buffer1.Complete();

}

Here是我使用的nuget包

EDIT 更新了上面的代码，在消费时将最小计数设为 10。至于你怀疑当我们倾斜分布时它是否符合要求，那么答案是 YES

【讨论】：

谢谢。我忘了提到消费者也应该阻塞直到缓冲区1中至少有10个项目。我已经更新了 Q。
那么不要使用BufferBlock 使用BatchBlock`，然后你应该完成
当我说 22 行不能平均分成 10 块和 5 块时，这是否也适用？
另外一个值得注意的有趣点是，您只需增加MaxDegreeofParallelism 属性的值就可以使这个过程并行执行。这样做时要小心，因为它会增加 CPU 的负载。您必须在并行处理和硬件之间取得平衡。
如果未完成的任务不应该是系统/2 上的处理器，那不是最大吗？

【解决方案2】：

我知道这个问题已经得到解答，但只是想通过实现自己的 GetConsumingEnumerable 版本来展示另一种方法。通过将集合作为参数传递来调用此方法而不是 BlockingCollection 方法：

    private IEnumerable<T> GetConsumingEnumerable<T>(BlockingCollection<T> sourceCollection)
    {
        var buffer = new List<T>();

        foreach (var item in sourceCollection.GetConsumingEnumerable())
        {
            buffer.Add(item);

            if (buffer.Count >= sourceCollection.BoundedCapacity)
            {
                foreach (var bItem in buffer)
                {
                    yield return bItem;
                }

                buffer.Clear();
            }
        }

        foreach (var bItem in buffer)
        {
            yield return bItem;
        }
    }

【讨论】：