Czy mogę ograniczyć ilość obiektów System.Threading.Tasks.Task, które działają jednocześnie?

Question

Używam klas zadań do wykonywania obliczeń wielowątkowych.

Na przykład:

List<Task> taskList = new List<Task>(); 
    for(int i=0;i<10;i++){ 
     var task = new Task(() => Console.WriteLine("Hello from taskA.")); 
     taskList.Add(task); 
     task.Start(); 
    } 

Czy istnieje sposób, aby tylko, powiedzmy, 3 Zadania uruchomić co najwyżej, a reszta czekać?

Answer 1

zmienić właściwość MaxDegreeOfParallelism.

example

Answer 2

polecam sprawdzenie słupek here.

Parafrazując, tworzysz kolekcję odpowiednich zadań, a następnie używasz ParallelOptions.MaxDegreeOfParallelism, aby kontrolować liczbę wykonań w tym samym czasie.

Answer 3

My blog post pokazuje, jak to zrobić zarówno z Zadaniami, jak iz Działaniami, i dostarcza przykładowy projekt, który można pobrać i uruchomić, aby zobaczyć oba w akcji.

z działaniami

przypadku korzystania z Actions, można skorzystać z wbudowanego w .NET Parallel.Invoke funkcję. Tutaj ograniczamy to do równoległego działania maksymalnie 3 wątków.

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

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

z zadaniami

Ponieważ używasz Zadania tu jednak nie ma wbudowaną funkcję. Możesz jednak użyć tego, który udostępniam na swoim blogu.

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

a następnie tworząc swoją listę zadań i wywołanie funkcji, aby je uruchomić, a powiedzieć maksymalnie 3 równolegle na raz, można to zrobić:

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

Answer 4

Można to wykorzystać jeden jako ogólny

public TaskThrottle(int initialCount, int maxTasksToRunInParallel) 
{ 
    _semaphore = new SemaphoreSlim(initialCount, maxTasksToRunInParallel); 
} 

public void TaskThrottler<T>(IEnumerable<Task<T>> tasks, int timeoutInMilliseconds, CancellationToken cancellationToken = default(CancellationToken)) where T : class 
{ 
     // Get Tasks as List 
     var taskList = tasks as IList<Task<T>> ?? tasks.ToList(); 
     var semaphoreTasks = new List<Task<int>>(); 

     // When the first task completed, flag as done/release 
     taskList.ForEach(x => 
     { 
      semaphoreTasks.Add(x.ContinueWith(y => _semaphore.Release(), cancellationToken)); 
     }); 

     semaphoreTasks.ForEach(async x => 
     { 
      // It will not pass this until one free slot available or timeout occure 
      if(timeoutInMilliseconds > 0) 
       await _semaphore.WaitAsync(timeoutInMilliseconds, cancellationToken); 
      else 
       await _semaphore.WaitAsync(cancellationToken); 

      // Throws a OperationCanceledException if this token has had cancellation requested 
      cancellationToken.ThrowIfCancellationRequested(); 

      // Start the task 
      x.Start(); 
     }); 

     Task.WaitAll(semaphoreTasks.ToArray(), cancellationToken); 
}