c# - 具有有限容量的转换 block 中的 TPL 数据流异常

Question

我需要构建将处理大量消息的 TPL 数据流管道。因为有很多消息，我不能简单地将它们 Post 到 BufferBlock 的无限队列中，否则我将面临内存问题。所以我想使用 BoundedCapacity = 1 选项来禁用队列并使用 MaxDegreeOfParallelism 来使用并行任务处理，因为我的 TransformBlock 可能需要一些时间对于每条消息。我还使用 PropagateCompletion 完成所有操作，并在管道中传播失败。

但是当错误发生在第一条消息之后时，我面临错误处理的问题:调用 await SendAsync 只是将我的应用程序切换到无限等待状态。

我已将案例简化为示例控制台应用程序:

var data_buffer = new BufferBlock<int>(new DataflowBlockOptions
{
    BoundedCapacity = 1
});

var process_block = new ActionBlock<int>(x =>
{
    throw new InvalidOperationException();
}, new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 2,
    BoundedCapacity = 1
});

data_buffer.LinkTo(process_block,
    new DataflowLinkOptions { PropagateCompletion = true });

for (var k = 1; k <= 5; k++)
{
    await data_buffer.SendAsync(k);
    Console.WriteLine("Send: {0}", k);
}

data_buffer.Complete();

await process_block.Completion;

Answer 1

这是预期的行为。如果“下游”存在故障，则错误不会向上“向后”传播到网格。网格期待您检测到该故障(例如，通过 process_block.Completion)并解决它。

如果你想向后传播错误，你可以在 process_block.Completion 上有一个 await 或继续 faults如果下游 block 出现故障，则上游 block 。

请注意，这不是唯一可能的解决方案；您可能想要重建网格的那部分或将源链接到替代目标。源 block 没有故障，因此它们可以继续处理修复后的网格。