具有并行性和有限容量延迟消息行为的转换块

Question

当TransformBlock 具有非无限的MaxDegreeOfParallelism > 1 和BoundedCapacity 时，为什么尽管输入队列中有容量，但在有一个长时间运行的任务时它会推迟接收更多消息？

采取以下控制台应用程序。它使用MaxDegreeOfParallelism = 5 和BoundedCapacity = 5 创建一个TransformBlock，然后向它提供100 条消息。当块处理消息x == 50 时，它会将该任务延迟 10 秒。

TransformBlock<int, string> DoSomething = new TransformBlock<int, string>(async (x) => {
    if (x == 50)
    {
        Console.WriteLine("x == 50 reached, delaying for 10 seconds.");
        await Task.Delay(10000);
    }
    Console.WriteLine($"processed message {x}");
    return x.ToString();
}, new ExecutionDataflowBlockOptions { BoundedCapacity = 5, MaxDegreeOfParallelism = 5 });

DoSomething.LinkTo(DataflowBlock.NullTarget<string>()); // ensure we empty the transform block

for (int i = 0; i < 100; i++)
{
    Stopwatch blockedTime = Stopwatch.StartNew();
    await DoSomething.SendAsync(i).ConfigureAwait(false);
    blockedTime.Stop();
    Console.WriteLine($"Submitted {i}\tBlocked for {blockedTime.ElapsedMilliseconds}ms.");
}

DoSomething.Complete();
await DoSomething.Completion;
Console.WriteLine("Completed.");
Console.ReadKey();

结果显示消息 50-54 都被区块接收。消息 51-54 已完成，然后控制台窗口在 10 秒内没有显示任何输出，然后才显示消息 50 已完成并且块能够接收到消息 55。

...
Submitted 50    Blocked for 0ms.
Submitted 51    Blocked for 0ms.
processed message 51
Submitted 52    Blocked for 0ms.
x == 50 reached, delaying for 10 seconds.
processed message 52
processed message 53
Submitted 53    Blocked for 0ms.
Submitted 54    Blocked for 0ms.
processed message 54 // when run, 10 seconds pause happens after displaying this line
processed message 50 
processed message 55
Submitted 55    Blocked for 9998ms.
...

为什么Transform Block 不继续将块填充到 Bounded Capacity 5，而使用其他 4 度的并行度继续处理消息？

ActionBlock 不显示这些症状并继续处理其他可用并行线路上的消息。

无限容量TransformBlock 也不会显示这些症状。

Answer 1

因为默认参数EnsureOrdered是true，所以它试图保持结果的顺序。也就是说，它无法继续处理超过BoundedCapacity，因为它需要维持秩序，这是您在测试中看到的背压。

此外，ActionBlock 不会表现出这种行为，因为它不会输出到任何其他块（可以说这是一个死胡同），因此没有 排序的概念 em>，背压只受限于有界容量和并行度。

DataflowBlockOptions.EnsureOrdered Property

默认情况下，数据流块对 消息。将 EnsureOrdered 设置为 false 会告诉一个块，如果它能够这样做，它可以放宽这个排序。 如果立即生成处理结果，这将是有益的 可用比保持输入到输出更重要 订购。

解决方法是删除已订购的要求

new ExecutionDataflowBlockOptions 
       { 
           BoundedCapacity = 5,
           MaxDegreeOfParallelism = 5 ,
           EnsureOrdered = false
       });