c# - 如何使用具有负载平衡和有限并行度的任务并行库 (TPL)？

Question

我的任务是使用(异步)接口(interface)将已知数量的值写入外部系统。我必须限制并发执行的最大并行写入数。此外，我必须使用负载平衡，因为该外部系统写入某些值可能需要更长的时间。

我知道如何单独解决这些问题:

并行度:

new ParallelOptions {MaxDegreeOfParallelism = maxNrParallelWrites}

我也偶然发现了这篇文章:http://msdn.microsoft.com/en-us/library/ee789351(v=vs.110).aspx

负载平衡:

var partitioner = Partitioner.Create(values.ToList(), true);

来自异步接口(interface)的任务:

var writeTask = Task<AccessResult>.Factory.FromAsync(BeginWriteValue, EndWriteValue, value.SystemId, value.Xml, priority, null);

但是我如何正确地结合所有这些技术呢？我创建了以下代码:

  int maxNrParallelWrites = GetMaxNrParallelWrites();
  var partitioner = Partitioner.Create(values.ToList(), true);
  Parallel.ForEach(partitioner, new ParallelOptions {MaxDegreeOfParallelism = maxNrParallelWrites},
    (val) =>
    {
      var writeValueTask = GetWriteValueTask(val, priority);
      Task.WaitAny(writeValueTask);
    });

我对前面代码的最后一部分特别不确定:执行工作负载的操作。直接使用同步接口(interface)而不是创建 WriteValueTask 会更好吗:

(val) =>
    {
      var accessResult = externalSystem.WriteValue(....);
    }

或者创建一个任务然后直接等待它(Task.WaitAny(...))是否可以？

Answer 1

您应该使用 TPL Dataflow 的 ActionBlock，它为您封装了所有内容。它是一个基于参与者的框架，是 TPL 的一部分:

var block = new ActionBlock<Value>(
    value => GetWriteValueTask(value, priority)
    new ExecutionDataflowBlockOptions()
    {
        MaxDegreeOfParallelism = GetMaxNrParallelWrites();
    });

foreach (var value in values)
{
    block.Post(value);
}

您可以设置 MaxDegreeOfParallelism、BoundedCapacity 并且加载平衡是内置的，因为它一次只处理 MaxDegreeOfParallelism 项，并且当每个完成它处理下一个(而不是使用预先对集合进行分区的 Partitioner)

注意:当您执行 async 任务并等待它同步完成(即 Task.WaitAny)时，实际上没有什么是异步的。在这种情况下，您应该使用 Task.WhenAny。