以无限并行性多次运行同一个 ActionBlock

Question

我正在学习 TPL 数据流，我通过我的一些朋友看到了它的强大功能，但我的实施遇到了问题。

我想要/需要的是尽可能快地发送消息。我正在 LinqPad 中做一些原型设计，这就是我目前所拥有的：

// Holds all the messages for my loadMessage ActionBlock to grab its data from
var bufferBlock = new BufferBlock<string>();

// Sends message to where it needs to go as fast as it can.
var loadMessage = new ActionBlock<string>(msg => 
{
    msg.Dump();
},  
new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
});

// Links the blocks together
bufferBlock.LinkTo(loadMessage);

// Loads the Buffer
for (int i = 0; i < 10; i++)
{
    bufferBlock.Post("This is a message");
}

//Calls completion to stop threads
bufferBlock.Complete();
loadMessage.Complete(); 

问题是 loadMessageBlock 没有在上面的示例中转储消息。我一直在寻找一些见解，但运气不佳。我认为我遗漏了 TPL 的一些基本内容。我的理解是 BufferBlock 保存要由其他块处理的信息，而 ActionBlocked（与 BufferBlock 链接）应该从缓冲区中获取数据并做它需要做的事情。在将信息放在缓冲区上的 For 循环停止完成后，将调用停止线程。

在我的实现中，我有一个 Parallel.For 可以很好地运行我的 loadMessage 中的代码。我只是无法实现 TPL 来做我想做的事，我的理解是 TPL 会比 Parallel.For 更快。

我认为这应该如何工作？我是否误用了 TPL？我将继续研究答案，任何指针将不胜感激。谢谢！

Answer 1

首先，关于术语的说明：TPL（Task Parallel Library 的缩写）与 TPL 数据流不同，它只是一个子集。整个 TPL 包括 Parallel.For() 和 Task 类型。

现在，您的代码的问题在于您过早地完成了loadMessage 块。在您拨打Complete() 后，该块将不再接受任何消息，因此您发布到bufferBlock 的消息将永远不会到达loadMessage。

您只需要在bufferBlock 将其所有消息发送给它之后完成loadMessage。这正是PropagateCompletion 所做的：

bufferBlock.LinkTo(
    loadMessage, new DataflowLinkOptions { PropagateCompletion = true });

// post your data to bufferBlock here

bufferBlock.Complete();
await loadMessage.Completion;

此外，在这种特定情况下，根本不需要bufferBlock，您可以直接将消息发布到loadMessage。

我的理解是 TPL 会比 Parallel.For 更快

我不明白为什么总体上应该更快。在正常情况下，它们的性能应该是相当的。因此，您应该使用更适合您的问题的一种，而不是因为“它更快”而选择一种。如果你真的那么在意性能，那就两种方式编写代码，然后衡量哪个更好。

Answer 2

1. 我想要/需要的是尽可能快地发送消息：

   为此，您需要同时向缓冲区块发送/接收数据。下面是sn-p：

   var bufferBlock = new BufferBlock<string>();
   // Write to and read from the message block concurrently. 
   var post01 = Task.Run(() =>
   {
       // Loads the Buffer
       for (int i = 0; i < 10; i++)
       {
           bufferBlock.Post(string.Format("This is a message {0}",i));
       }
   });
   var receive = Task.Run(() =>
   {
       for (int i = 0; i < 10; i++)
       {
           var message = bufferBlock.Receive();
           message.Dump();
       }
   });
   
   
   Task.WaitAll(post01, receive);
   

   更多信息请访问MSDN link

2. 我的理解是 TPL 会比 Parallel.For 更快。

   这不正确，因为它们使用相同的底层结构。它们属于同一个命名空间System.Threading.Tasks