c# - TPL数据流如何与 "global"数据同步

Question

我实际上是在学习 TPL 数据流。每当我读到有关它的内容时，我都觉得 ok 听起来不错，但后来我经常问自己:“好吧，如果我有一个管理器来处理不同的 session 怎么办。这些 session 可以通过某些消息进行更新。如果复杂的 TPL 数据流网格我必须内置同步机制来访问管理器，这会减慢或阻塞网格。”

为 TPL 数据流使用管理对象感觉有点不对。

谁能给我一些“正确”方向的提示(链接、书籍、示例...)，如何解决上述示例。

Answer 1

没有一些代码的非常广泛的问题。但通常你要么通过 Tuple 或一些 DTO 传递一些“状态”对象，我个人认为这是泄漏设计，要么你可以将一些 block 注入(inject)你的管道。

例如，您可以创建一个 WriteOnceBlock为它提供一个 session 值，对于每个管道都是唯一的，并简单地通过执行期间发生的新事件通知它的值。如果您为不同的 session 创建管道，这可能是您的一个选择，但如果您有一个大管道，则需要另一种方法。

例如，您有一个 BufferBlock , 一个 ActionBlock ，执行 session 更新，以及 TransformBlock ，它只是继续您的正常管道执行。在这种情况下你可以做的是引入一个 BroadcastBlock ，将其与 Action block 和变换 block 链接。现在，您使用谓词将缓冲区链接到广播 block ，然后将缓冲区直接链接到转换 block 。

这里的想法是，如果您的消息被过滤(因此需要 session 更新)，它将进入广播 block ，然后进入 session 更新操作和您的正常执行管道。如果它与谓词不匹配，它会简单地检查您的正常工作流程。

PS:如果这在文字上太复杂，我可以为这种情况提供一些示例代码。

var linkOptions = new DataflowLinkOptions { PropagateCompletion = true };

var buffer = new BufferBlock<int>();
// broadcast do copy according lambda from constructor
// if you provide same reference for message, make sure that your access is thread-safe
var broadcast = new BroadcastBlock<int>(i => i);
buffer.LinkTo(broadcast, linkOptions);

// session update block
var sessionUpdate = new ActionBlock<int>(i =>
    {
        Thread.Sleep(new Random().Next(1000));
        Console.WriteLine($"Session update:{i}");
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });
// normal execution
var transform = new TransformBlock<int, int>(i =>
    {
        Thread.Sleep(new Random().Next(1000));
        Console.WriteLine($"Normal execution:{i}");
        return i;
    }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });
// do not complete the standalone session block
// message will be accepted only for multipliers of 3
broadcast.LinkTo(sessionUpdate, i => i % 3 == 0);
// normal pipeline with completion propagation
broadcast.LinkTo(transform, linkOptions);

for (var i = 0; i < 10; ++i)
{
    // async message
    await buffer.SendAsync(i);
}
buffer.Complete();
await transform.Completion;