System.Threading.Tasks - 限制并发任务的数量

Question

我刚刚开始着眼于.Net 4.0中的新“System.Threading.Tasks”善良，并想知道是否有任何构建支持限制一次运行的并发任务的数量，或者是否应该手动处理。 E：如果我需要调用100次计算方法，有没有办法设置100个任务，但只有5个同时执行？答案可能只是创建5个任务，调用Task.WaitAny，并在每个任务完成时创建一个新任务。如果有更好的方法可以做到这一点，我只想确保我不会错过任何一招。

基本上是有一个内置的方式做到这一点：

Dim taskArray() = {New Task(Function() DoComputation1()), 
        New Task(Function() DoComputation2()), 
        ... 
        New Task(Function() DoComputation100())} 

Dim maxConcurrentThreads As Integer = 5 
RunAllTasks(taskArray, maxConcurrentThreads) 

感谢您的帮助。

Answer 1

我知道这几乎是一岁，但我已经找到了一个更简单的方法来做到这一点，所以我想我也有同感：

Dim actionsArray() As Action = 
    new Action(){ 
     New Action(Sub() DoComputation1()), 
     New Action(Sub() DoComputation2()), 
     ... 
     New Action(Sub() DoComputation100()) 
     } 

System.Threading.Tasks.Parallel.Invoke(New Tasks.ParallelOptions() With {.MaxDegreeOfParallelism = 5}, actionsArray) 

瞧！

Answer 2

虽然您可以创建实现此类行为的TaskScheduler的子类，但它看起来并不像它。

Answer 3

简短回答：如果你想要的是限制工作任务的数量，使他们不会饱和你的web服务，那么我认为你的方法是好的。

Long答案： .NET 4.0中的新System.Threading.Tasks引擎运行在.NET ThreadPool之上。由于每个进程只有一个ThreadPool，并且默认最多有250个工作线程。因此，如果要将ThreadPool的最大线程数设置为更适中的数，则可以减少并发执行的线程数，从而减少使用ThreadPool.SetMaxThreads (...) API的任务数。然而，请注意，您可能不会单独使用ThreadPool，因为您使用的许多其他类也可能会将项目排入ThreadPool。因此，这样做很可能最终导致应用程序的其他部分崩溃。还要注意，因为ThreadPool使用一种算法来优化对给定机器底层内核的使用，所以限制线程池可以排列到任意低数量的线程数可能会导致一些相当严重的性能问题。同样，如果你想执行少量的工作任务/线程来执行一些任务，那么只有创建少量任务（对比100）是最好的方法。

Answer 4

我知道这是一个旧线程，但我只是想分享我的解决方案：使用信号量。

（这是在C＃）

private void RunAllActions(IEnumerable<Action> actions, int maxConcurrency) 
{ 
    using(SemaphoreSlim concurrencySemaphore = new SemaphoreSlim(maxConcurrency)) 
    { 
     foreach(Action action in actions) 
     { 
      Task.Factory.StartNew(() => 
      { 
       concurrencySemaphore.Wait(); 
       try 
       { 
        action(); 
       } 
       finally 
       { 
        concurrencySemaphore.Release(); 
       } 
      }); 
     } 
    } 
} 

Answer 5

一个解决办法可能是先来看看微软here预先做好的代码。

描述是这样的：“提供一个任务调度程序，确保在ThreadPool上运行时具有最大的并发级别”，并且就我已经能够测试它而言，似乎可以做到这一点，与ParallelOptions中的MaxDegreeOfParallelism属性相同。

Answer 6

如果您的程序使用webservices，同时连接数将限制为ServicePointManager.DefaultConnectionLimit属性。如果你想要5个同时连接，使用Arrow_Raider的解决方案是不够的。你也应该增加ServicePointManager.DefaultConnectionLimit，因为它默认只有2。

Answer 7

C＃等效采样由詹姆斯

Action[] actionsArray = new Action[] { 
new Action(() => DoComputation1()), 
new Action(() => DoComputation2()), 
    //... 
new Action(() => DoComputation100()) 
    }; 

    System.Threading.Tasks.Parallel.Invoke(new Tasks.ParallelOptions {MaxDegreeOfParallelism = 5 }, actionsArray) 

Answer 8

My blog post提供了如何都与任务，并以行动做到这一点，并提供您可以下载并运行同时看到在行动的示例项目。

随着使用操作操作

如果，您可以使用内置的.Net Parallel.Invoke功能。这里我们限制它并行运行最多5个线程。

var listOfActions = new List<Action>(); 
for (int i = 0; i < 100; i++) 
{ 
    // Note that we create the Action here, but do not start it. 
    listOfActions.Add(() => DoSomething()); 
} 

var options = new ParallelOptions {MaxDegreeOfParallelism = 5}; 
Parallel.Invoke(options, listOfActions.ToArray()); 

随着任务

由于您使用此任务虽然没有内置的功能。但是，您可以使用我在我的博客上提供的那个。

/// <summary> 
    /// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel. 
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para> 
    /// </summary> 
    /// <param name="tasksToRun">The tasks to run.</param> 
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param> 
    /// <param name="cancellationToken">The cancellation token.</param> 
    public static void StartAndWaitAllThrottled(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken()) 
    { 
     StartAndWaitAllThrottled(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken); 
    } 

    /// <summary> 
    /// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel. 
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para> 
    /// </summary> 
    /// <param name="tasksToRun">The tasks to run.</param> 
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param> 
    /// <param name="timeoutInMilliseconds">The maximum milliseconds we should allow the max tasks to run in parallel before allowing another task to start. Specify -1 to wait indefinitely.</param> 
    /// <param name="cancellationToken">The cancellation token.</param> 
    public static void StartAndWaitAllThrottled(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken()) 
    { 
     // Convert to a list of tasks so that we don&#39;t enumerate over it multiple times needlessly. 
     var tasks = tasksToRun.ToList(); 

     using (var throttler = new SemaphoreSlim(maxTasksToRunInParallel)) 
     { 
      var postTaskTasks = new List<Task>(); 

      // Have each task notify the throttler when it completes so that it decrements the number of tasks currently running. 
      tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release()))); 

      // Start running each task. 
      foreach (var task in tasks) 
      { 
       // Increment the number of tasks currently running and wait if too many are running. 
       throttler.Wait(timeoutInMilliseconds, cancellationToken); 

       cancellationToken.ThrowIfCancellationRequested(); 
       task.Start(); 
      } 

      // Wait for all of the provided tasks to complete. 
      // We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler&#39;s using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object. 
      Task.WaitAll(postTaskTasks.ToArray(), cancellationToken); 
     } 
    } 

，然后创建您的任务列表，并调用函数让他们跑，有说在同一时间最多5个同时的，你可以这样做：

var listOfTasks = new List<Task>(); 
for (int i = 0; i < 100; i++) 
{ 
    var count = i; 
    // Note that we create the Task here, but do not start it. 
    listOfTasks.Add(new Task(() => Something())); 
} 
Tasks.StartAndWaitAllThrottled(listOfTasks, 5);