TPL 数据流 - 条件循环

Question

目前我正在处理管道数据流，其中除了第 1 阶段之外的每个阶段都是async 正在运行的消费者和生产者。我有对象“流动”通过我的管道，这些对象引用了项目。在第 3 阶段，我想创建一个循环并缓冲所有满足特殊条件的对象（阶段循环）。

如果新对象进入（第 3 阶段）而当前有其他对象缓冲（阶段循环），我想检查它们是否在其引用项中匹配，如果匹配，则将它们发布到阶段循环的 BufferBlock。

问题是，如何从 Stage 3 中检查 Stage Loop 中所有对象的引用项？

管道有点像这样：

Incoming objects ->  
  BufferBlock1 -> Parsing (Stage2) ->  
  BufferBlock2 -> Processing (Stage3) ->
  BufferBlock3 -> Stage Loop ->  
    Back to BufferBlock 2

Answer 1

您的链中确实不需要那么多BufferBlock。 TPL Dataflow 包含一个 TransformBlock，它封装了 BufferBloсk 和 ActionBlock 逻辑，并有一个用于处理消息的输出块。

至于循环，您可以将块相互链接with static extension method，所以这可能是这样的

stage2.LinkTo(stage3, CheckForExistingProcessing);
stage2.LinkTo(stage4);

Jere stage4 是未通过检查的消息队列，必须循环处理。您可以设置额外的ActionBlock，或者，也许，只需使用TransformBlock 将消息再次发送到适当的阶段。我认为您也可以引入重试检查，因为某些消息可能根本无法处理，因此有些原因。

另外，正如你所说，你有async 逻辑，你可能应该SendAsync 消息而不是Post 它们（你也可以使用CancellationToken 的重载）：

// asynchronously wait for a sending with resending attempts
await stage1.SendAsync(m);
// asynchronously wait for a sending with resending attempts with possible cancellation
await stage2.SendAsync(m, token);

Post 方法是同步的，如果目标不接受消息，丢弃消息，比较 SendAsync 方法，即使目标现在不能接受它也会尝试传递消息。 p>