带有PropagateCompletion 配置的LinkTo 方法将源块的完成传播到目标块。因此，如果源块失败，失败将传播到目标块，因此最终两个块都会完成。如果目标块失败，则情况并非如此。在这种情况下，源块将不会收到通知，并将继续接受和处理消息。如果我们在混合中添加BoundedCapacity 配置，源块的内部输出缓冲区很快就会变满，从而阻止它接受更多消息。正如您所发现的，这很容易导致死锁。
为了防止发生死锁，最简单的方法是确保管道的任何块中的错误都会导致其所有组成块尽快及时完成。正如 Stephen Cleary 的 answer 所指出的，其他方法也是可能的，但在大多数情况下，我希望快速失败的方法是理想的行为。令人惊讶的是，这种简单的行为并不那么容易实现。没有现成的内置机制可用于此目的，并且手动实现它很棘手。
从 .NET 6 开始，强制完成作为数据流管道一部分的块的唯一可靠方法是 Fault 块，并通过将其链接到 NullTarget 来丢弃其输出缓冲区。仅故障块，或通过CancellationToken 选项取消它是不够的。在某些情况下，故障或取消的块将无法完成。 Here 是第一种情况的演示（故障且未完成），here 是第二种情况的演示（已取消且未完成）。这两种情况都要求该块先前已标记为已完成，这对于参与数据流管道的所有块可以自动且不确定地发生，并与PropagateCompletion 配置相关联。存在报告此问题行为的 GitHub 问题：No way to cancel completing dataflow blocks。截至撰写本文时，开发人员尚未提供任何反馈。
有了这些知识，我们可以实现一个LinkTo-on-steroids 方法，该方法可以创建如下所示的快速失败管道：
/// <summary>
/// Connects two blocks that belong in a simple, straightforward,
/// one-way dataflow pipeline.
/// Completion is propagated in both directions.
/// Failure of the target block causes purging of all buffered messages
/// in the source block, allowing the timely completion of both blocks.
/// </summary>
/// <remarks>
/// This method should be used only if the two blocks participate in an exclusive
/// producer-consumer relationship.
/// The source block should be the only producer for the target block, and
/// the target block should be the only consumer of the source block.
/// </remarks>
public static async void ConnectTo<TOutput>(this ISourceBlock<TOutput> source,
ITargetBlock<TOutput> target)
{
source.LinkTo(target, new DataflowLinkOptions { PropagateCompletion = true });
try { await target.Completion.ConfigureAwait(false); } catch { }
if (!target.Completion.IsFaulted) return;
if (source.Completion.IsCompleted) return;
source.Fault(new Exception("Pipeline error."));
source.LinkTo(DataflowBlock.NullTarget<TOutput>()); // Discard all output
}
使用示例：
var data_buffer = new BufferBlock<int>(new() { BoundedCapacity = 1 });
var process_block = new ActionBlock<int>(
x => throw new InvalidOperationException(),
new() { BoundedCapacity = 2, MaxDegreeOfParallelism = 2 });
data_buffer.ConnectTo(process_block); // Instead of LinkTo
foreach (var k in Enumerable.Range(1, 5))
if (!await data_buffer.SendAsync(k)) break;
data_buffer.Complete();
await process_block.Completion;
您也可以考虑等待管道的所有组成块，之前等待最后一个（或在finally 区域中之后）。这样做的好处是，在发生故障时，在管道的下一次重生之前，您不会冒着泄漏在后台运行的即发即弃操作的风险：
try { await Task.WhenAll(data_buffer.Completion, process_block.Completion); } catch { }
您可以忽略await Task.WhenAll 操作可能引发的所有错误，因为等待最后一个块无论如何都会传达大部分与错误相关的信息。您可能只会错过下游块失败后上游块中发生的其他错误。如果需要，您可以尝试观察所有错误，但这会很棘手，因为错误是如何向下游传播的：您可能会多次观察到相同的错误。如果您想认真记录每一个错误，在处理块的 lambdas 中进行记录可能更容易（也更准确），而不是依赖于它们的 Completion 属性。
缺点：上面的ConnectTo 实现一次将故障向后传播一个块。传播不是即时的，因为在任何当前处理的消息的处理完成之前，故障块不会完成。如果管道很长（5-6 个块或更多），并且每个块的工作量很大，这可能是一个问题。这种额外的延迟不仅浪费时间，而且浪费资源，因为这些工作无论如何都会被丢弃。
我在this GitHub 存储库中上传了一个更复杂的ConnectTo 想法版本。它解决了上一段中提到的延迟完成问题：任何块中的失败都会立即传播到所有块。作为奖励，它还会传播管道中的所有错误，作为平面 AggregateException。
注意：此答案已从头开始重写。最初的答案 (Revision 4) 包含一些错误的想法，以及 ConnectTo 方法的有缺陷的实现。