如何使用 TPL 数据流处理完整的输入列表？答案

【问题标题】：How to process a complete list of inputs with TPL Dataflow?如何使用 TPL 数据流处理完整的输入列表？
【发布时间】：2019-07-29 16:39:35
【问题描述】：

我是 TPL Dataflow 的新手，我可以使用它，但我不确定我是否正确使用它。我有一个输入（字符串）列表，我想以最大程度的并行度处理它们（全部），并知道它什么时候完成。现在我只是通过输入foreach 并在ActionBlock 上调用Post，忽略返回值。这似乎不正确，因为它可能会丢失输入。

我的问题是：如何避免丢失物品？是否有一个内置块，我可以只给我的输入，它会确保它们都被尝试？（无论每次输入成功/失败。）

我看到的建议基本上是：

await block.Completion;

这是否会导致输入失败（Post 或 SendAsync 将返回 false）？对我来说奇怪的是，似乎这个决定是在我打电话给Post 而不是之后做出的，所以这个Completion 甚至不包括这些项目。

我觉得我基本上需要一个重试循环来处理以前无法处理的输入，类似于：

while (items.Count > 0) {
  foreach (var item in items) {
    if (await block.SendAsync(item)) {
      items.Remove(item);
    }
  }

  await block.Completion;
}

block.Complete();

（除了更好的循环处理/错误检查。）

这个额外的级别是不必要的吗？还是我在概念上的某个地方错了？

【问题讨论】：

相关：TPL Dataflow, whats the functional difference between Post() and SendAsync()?

标签： c# tpl-dataflow

【解决方案1】：

这似乎不正确，因为它可能会丢失输入。

假设您使用默认值，这是正确的。 Post 仅在块拒绝输入时返回 false。如果块接收到Complete 信号，或者块的输入缓冲区已满，则可能会发生这种情况。默认情况下，每个块的输入缓冲区可以无限增长，因此具有默认输入缓冲区大小的ActionBlock 只会在调用Complete 后从Post 返回false。

ActionBlock 最常见的用例是使用 unlimited bounded capacity，并且代码仅在添加所有项目后调用 Complete。在这种情况下，Post 永远不会返回 false，您可以放心地忽略返回值。

【讨论】：

【解决方案2】：

如果块已完成或块的输入缓冲区已满，Post 方法将返回 false。由于设置BoundedCapacity 不是什么奇特的东西，并且很可能在项目的后期阶段需要解决内存使用率高的新问题，我认为使用Post 方法并不安全，只是简单地忽略结果。为了避免出现涉及丢失消息（可能是订单或发票）的有趣错误，您可以执行以下操作：

foreach (var item in items)
{
    var accepted = block.Post(item);
    if (!accepted) throw new InvalidOperationException("Item was not accepted");
}

这样你至少会被告知有什么东西坏了，并且不会让错误的行为潜入。

另一方面，等待SendAsync 并忽略结果要安全得多。在发生异常或发生取消的情况下，SendAsync 通常会返回false，在这种情况下，您将在await 处收到有关块的Completion 的通知。所以在这种情况下不需要抛出异常。

foreach (var item in items)
{
    await block.SendAsync(item).ConfigureAwait(false);
}

出于性能原因，您可以同时使用Post 和SendAsync。只有当您有数千万个项目要处理时，这才会有所作为。

foreach (var item in items)
{
    if (!block.Post(item))
    {
        await block.SendAsync(item).ConfigureAwait(false);
    }
}

【讨论】：