c# - 意外行为 - TPL DataFlow BatchBlock 在 TriggerBatch 执行时拒绝项目

Question

当您创建具有有限容量的 batchblock 并在(并行地)发布新项目时调用 triggerBatch - 发布新项目将在触发器批处理执行期间失败。

调用触发器批处理(每 X 次)是为了确保数据不会在 block 中延迟太久，以防传入数据流暂停或减慢。

以下代码将输出一些“失败后”事件。例如:

    public static void Main(string[] args)
    {
        var batchBlock = new BatchBlock<int>(10, new GroupingDataflowBlockOptions() { BoundedCapacity = 10000000 });
        var actionBlock = new ActionBlock<int[]>(x => ProcessBatch(x), new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 });
        batchBlock.LinkTo(actionBlock);

        var producerTask = Task.Factory.StartNew(() =>
        {
            //Post 10K Items
            for (int i = 0; i < 10000; i++)
            {
                var postResult = batchBlock.Post(i);
                if (!postResult)
                    Console.WriteLine("Failed to Post");
            }
        });

        var triggerBatchTask = Task.Factory.StartNew(() =>
            {                    
                //Trigger Batch..
                for (int i = 0; i < 1000000; i++)
                    batchBlock.TriggerBatch();
            });

        producerTask.Wait();
        triggerBatchTask.Wait();
    }

    public static void ProcessBatch(int[] batch)
    {
        Console.WriteLine("{0} - {1}", batch.First(), batch.Last());
    }

*请注意，只有当 batchBlock 有界时，这种情况才可重现。

我是不是遗漏了什么或者是 batchBlock 的问题？

Answer 1

BatchBlock 并没有真正拒绝该项目，它试图推迟它。除了在 Post() 的情况下，推迟不是一个选项。解决此问题的一个简单方法是使用 await batchBlock.SendAsync(i) 而不是 batchBlock.Post(i) (这也意味着您需要更改 Task.Factory.StartNew(() => 到 Task.Run(async () =>)。

为什么会这样？根据the source code ，如果 BatchBlock 是有界的，则 TriggerBatch() 会被异步处理，并且在处理期间不会接受任何新项目。

在任何情况下，您都不应该期望 Post() 在有界 block 上总是返回 true，如果 block 已满，Post( ) 也将返回 false。