不遵守链接的 ActionBlock 的 BoundedCapacity

Question

我有一个包含两个步骤的顺序管道。

（简化示例）

第一步只是将输入数字加 1000。 第二步只是显示数字。

var transformBlock = new TransformBlock<int, long>(StepOne, new ExecutionDataflowBlockOptions
{
      MaxDegreeOfParallelism = 1,
      BoundedCapacity = DataflowBlockOptions.Unbounded,
});
var actionBlock = new ActionBlock<long>(StepTwo, new ExecutionDataflowBlockOptions
{
      MaxDegreeOfParallelism = 1,
      BoundedCapacity = 2,
});
transformBlock.LinkTo(actionBlock, new DataflowLinkOptions
{
    PropagateCompletion = true
});

for (int i = 0; i < 100; i++)
{
    transformBlock.Post(i);
}

static async Task<long> StepOne(int item)
{
    await Task.Delay(500);
    Console.WriteLine("transforming: " + item);
    return (long)item + 1000;
}

static async Task StepTwo(long item)
{
    await Task.Delay(1000);
    Console.WriteLine("final product: " + item);
}

由于第 2 步花费的时间比第 1 步长，我预计第 1 步会在一段时间后进行节流，因为它无法将结果发送到第 2 步的有界缓冲区。

预期输出：
转型：0
转型：1
最终产品：1000
转型：2
最终产品：1001
转型：3
最终产品：1002
转型：4
最终产品：1003
...

实际输出：
转型：0
转型：1
最终产品：1000
转型：2
转型：3
最终产品：1001
转型：4
转型：5
最终产品：1002
转型：6
转型：7
最终产品：1003
...

Answer 1

TransformBlock 在内部维护两个队列，一个输入队列和一个输出队列。这两个队列的大小可以通过InputCount 和OutputCount 属性随时监控。这两个队列的累加大小由BoundedCapacity选项配置，所以总和InputCount+OutputCount总是小于等于BoundedCapacity的值。在您的情况下，块的BoundedCapacity 是Unbounded，因此对于这两个队列可以变得多大没有限制因素（可能除了像Int32.MaxValue 这样的一些硬限制）。链接的ActionBlock 具有有限的有界容量这一事实几乎是无关紧要的，除了延迟将转换后的值从TransformBlock 的输出队列传输到ActionBlock 的输入队列之外，没有任何后果。仅当您监视源块的OutputCount 属性和目标块的InputCount 属性时，才能观察到此结果。 TransformBlock 是否没有链接到任何目标块甚至都没有关系。它会很高兴地继续自己处理数字，直到达到某个硬限制，或者机器的内存耗尽。