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);