使用 TPL 做一些“工作”答案

【问题标题】：Using TPL to do some "Jobs"使用 TPL 做一些“工作”
【发布时间】：2011-04-15 19:09:51
【问题描述】：

我正在启动一个小“作业管理器”，它将执行一些“作业”对象实现“IJob”方法。每个 IJob 都通过返回一个潜在的新工作列表来结束。

我所有的作业都存储在一个队列中，该队列可以线程安全地访问，当我完成一个作业时，我会将新查询的作业添加到这个队列中。

我正在寻找如何将它与 TPL 并行化。

我有几个想法，但由于我的限制，它们都不是有效的，它们是：

我需要限制线程数（假设为 4），因为某些请求正在查询一个网站，这不允许超过 4 个同时请求（我想我可以用信号量来管理）
我的作业列表会改变，所以我不能启动 4 个线程，将作业数除以线程数，每个线程都运行他的作业堆栈。
也许有时候，如果我有 4 个线程，我只剩下一个作业，但我不能让其他线程停止，因为也许我要运行的最后一个作业会创建更多作业。

非常感谢！

【问题讨论】：

标签： multithreading parallel-processing task-parallel-library

【解决方案1】：

这是真正的动态任务并行。您的代码循环遍历您拥有的作业并执行每个作业。每个作业都可以使用 addMethod 和并发队列添加新作业。

public static void ParallelWhileNotEmpty<T>(
  IEnumerable<T> initialValues, 
  Action<T, Action<T>> body)
{
  var opts = new ParallelOptions { MaxDegreeOfParallelism = 10 };
  var from = new ConcurrentQueue<T>(initialValues);  
  while (!from.IsEmpty)
  { 
    var to = new ConcurrentQueue<T>();
    Action<T> addMethod = to.Enqueue; 
    Parallel.ForEach(from, opts. body(v, addMethod));        
    from = to;
  }
}

这样，“循环”是开放式的，并且会一直持续到您用完工作为止。显然，您的真实应用程序会考虑重复的 URL 而不会添加它们等。但这允许您的应用程序动态添加工作。您可以使用 ParallelOptions 来限制并发，也可以编写调度程序。

有关动态任务并行性的更多信息，请参阅

http://msdn.microsoft.com/en-us/library/ff963551.aspx

示例的完整代码见

http://parallelpatterns.codeplex.com/SourceControl/changeset/view/54510#795590

以上两个都讨论了这个主题的其他替代变体。

如果您想要自定义调度程序来限制并行度，请参阅 MSDN 上的示例

http://msdn.microsoft.com/en-us/library/ee789351.aspx

【讨论】：

【解决方案2】：

您可以简单地使用ParallelOptions.MaxDegreeOfParallelism 来限制同时执行多少并发任务。

Reed Copsey 在他的博客上有an example。

【讨论】：

是的，但我仍然有我的主要问题，这对于更改的集合不起作用。他们总是做一个 collection.AsParallel()，但这不会考虑我添加到集合中的所有新作业:(

【解决方案3】：

查看 Campbell、Johnson、Miller 和 Toub 所著的 Parallel Programming with Microsoft .NET ... 特别是第 3 章的“具有线程本地工作队列的全局队列，通过 TPL，采用工作窃取算法”来处理负载平衡.

我将此作为我的一个设计的基础，它确实在性能上产生了巨大的差异。

【讨论】：

是的，但我的问题并不是真正的“负载平衡”，更多的是当我要求开始工作时我没有完成所有任务。当我只有一个工作时，我看不出负载平衡有什么帮助，而且这个工作会产生很多未来的工作，4 个工作中有 3 个将停止：/