“有界” BatchBlock => ActionBlock。如何完成正确的方法？答案

【问题标题】："bounded" BatchBlock => ActionBlock. How to complete the proper way?“有界” BatchBlock => ActionBlock。如何完成正确的方法？
【发布时间】：2015-07-16 09:45:53
【问题描述】：

我正在尝试使用链接到操作块的有界批处理块。我知道批处理块中的项目何时结束并且我想触发完成链。

问题是：如果我的BatchBlock<T> 属于给定的BoundedCapacity，我不会在动作块中触发我的所有项目。

这是我的问题的一个示例，它应该（根据我对 TPL 数据流的理解......）打印 0 到 124，但它最终会打印 0 到 99。

一定有我遗漏的东西...也许BoundedCapacity 的意思是“当队列计数超过 xxx 时丢弃项目...”如果是这样，我怎样才能保证最大的内存消耗？

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks.Dataflow;

namespace ConsoleApplication
{
    class Program
    {
        static void Main(string[] args)
        {
            int itemsCount = 125;
            List<int> ints = new List<int>(itemsCount);
            for (int i = 0; i < itemsCount; i++)
                ints.Add(i);

            BatchBlock<int> batchBlock = new BatchBlock<int>(50,new GroupingDataflowBlockOptions(){BoundedCapacity = 100});
            ActionBlock<int[]> actionBlock = new ActionBlock<int[]>(intsBatch =>
            {
                Thread.Sleep(1000);
                foreach (int i in intsBatch)
                    Console.WriteLine(i);               
            });
            batchBlock.LinkTo(actionBlock, new DataflowLinkOptions() { PropagateCompletion = true });

            // feed the batch block
            foreach (int i in ints)
                batchBlock.Post(i);
            // Don't know how to end the proper way... Meaning it should display 0 to 124 and not 0 to 99
            batchBlock.Complete();
            batchBlock.TriggerBatch();
            actionBlock.Completion.Wait();
        }
    }
}

【问题讨论】：

标签： c# .net task-parallel-library tpl-dataflow

【解决方案1】：

Post 并不总是成功。它尝试向块发布消息，但如果到达BoundedCapacity，它将失败并返回false。

你可以做的是使用SendAsync 来代替它返回一个等待的任务。如果该块有空间容纳您的消息，它会异步完成。如果没有，则该块返回一个任务，该任务将在它确实有空间接受新消息时完成。您可以等待该任务并限制您的插入：

async Task MainAsync()
{
    var ints = Enumerable.Range(0, 125).ToList();
    var batchBlock = new BatchBlock<int>(50, new GroupingDataflowBlockOptions { BoundedCapacity = 100 });
    var actionBlock = new ActionBlock<int[]>(intsBatch =>
    {
        Thread.Sleep(1000);
        foreach (var i in intsBatch)
            Console.WriteLine(i);
    });
    batchBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });

    foreach (var i in ints)
        await batchBlock.SendAsync(i); // wait synchronously for the block to accept.

    batchBlock.Complete();
    await actionBlock.Completion;
}

【讨论】：