작업 병렬 라이브러리의 스레드 수 제한

Question

Azure Blob 저장소에 업로드해야하는 파일이 몇백 개 있습니다.
병렬 작업 라이브러리를 사용하고 싶습니다.
그러나 파일 목록에서 foreach로 업로드 할 100 개의 스레드를 모두 실행하는 대신 최대 스레드 수를 제한하여 병렬로 작업을 완료하고 완료 할 수는 있습니다. 또는 자동으로 균형을 조정합니까?

Answer 1

스레드를 사용해서는 안됩니다. 자연히 비동기 인 Task 기반의 API가 있습니다 : CloudBlockBlob.UploadFromFileAsync. 병렬 업로드 수를 줄이려면 async/await 및 SemaphoreSlim과 함께 사용하십시오.

예 (안된) :

기본적으로

const MAX_PARALLEL_UPLOADS = 5; 

async Task UploadFiles() 
{ 
    var files = new List<string>(); 
    // ... add files to the list 

    // init the blob block and 
    // upload files asynchronously 
    using (var blobBlock = new CloudBlockBlob(url, credentials)) 
    using (var semaphore = new SemaphoreSlim(MAX_PARALLEL_UPLOADS)) 
    { 
     var tasks = files.Select(async(filename) => 
     { 
      await semaphore.WaitAsync(); 
      try 
      { 
       await blobBlock.UploadFromFileAsync(filename, FileMode.Create); 
      } 
      finally 
      { 
       semaphore.Release(); 
      } 
     }).ToArray(); 

     await Task.WhenAll(tasks); 
    } 
} 

Answer 2

MaxDegreeOfParallelism을 사용해 보셨습니까? 이처럼 :

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

Answer 3

당신은이 작업을 실행하여 확인할 수 있습니다 :

class Program 
{ 
    static void Main(string[] args) 
    { 
     var list = new List<int>(); 

     for (int i = 0; i < 100; i++) 
     { 
      list.Add(i); 
     } 

     var runningIndex = 0; 

     Task.Factory.StartNew(() => Action(ref runningIndex)); 

     Parallel.ForEach(list, i => 
     { 
      runningIndex ++; 
      Console.WriteLine(i); 
      Thread.Sleep(3000); 
     }); 

     Console.ReadKey(); 
    } 

    private static void Action(ref int number) 
    { 
     while (true) 
     { 
      Console.WriteLine("worked through {0}", number); 
      Thread.Sleep(2900); 
     } 
    } 
} 

당신이 병렬의 수는 시작에 작은 볼 수 있듯이, 더 큰 가져오고 끝으로 작은 성장한다. 따라서 자동 최적화의 일종이 분명히 있습니다.

Answer 4

당신이 수를 제한, 그 목록을 각 파일 업로드에 대한 활동이나 작업을 만들 목록에 넣어, 다음 처리 할거야 병렬로 처리 할 수 ​​있습니다.

My blog post은 작업 및 작업 모두에서이 작업을 수행하는 방법을 보여주고 다운로드하여 실행할 수있는 샘플 프로젝트를 제공합니다. 작업을 사용하여 작업

하다면

, 당신은 내장 된 닷넷 Parallel.Invoke 기능을 사용할 수 있습니다. 여기에서는 병렬로 최대 5 개의 스레드를 실행하도록 제한합니다.

var listOfActions = new List<Action>(); 
foreach (var file in files) 
{ 
    var localFile = file; 
    // Note that we create the Task here, but do not start it. 
    listOfTasks.Add(new Task(() => blobBlock.UploadFromFileAsync(localFile, FileMode.Create))); 
} 

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

이 옵션은 UploadFromFileAsync의 비동기 특성을 사용하지 않으므로 아래의 작업 예제를 사용하는 것이 좋습니다.

작업 포함

작업에는 기본 제공 기능이 없습니다. 그러나, 내 블로그에서 제공하는 것을 사용할 수 있습니다.

/// <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 async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken()) 
    { 
     await StartAndWaitAllThrottledAsync(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken); 
    } 

    /// <summary> 
    /// Starts the given tasks and waits for them to complete. This will run the specified number of tasks in parallel. 
    /// <para>NOTE: If a timeout is reached before the Task completes, another Task may be started, potentially running more than the specified maximum allowed.</para> 
    /// <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 async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken()) 
    { 
     // Convert to a list of tasks so that we don'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. 
       await throttler.WaitAsync(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'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. 
      await Task.WhenAll(postTaskTasks.ToArray()); 
     } 
    } 

그리고 작업의 목록을 작성하고 한 번에 5 개 동시에 최대 말과 함께, 그들을 실행하도록 함수를 호출, 당신이 할 수 있습니다 : 당신은 사용하지 말아야

var listOfTasks = new List<Task>(); 
foreach (var file in files) 
{ 
    var localFile = file; 
    // Note that we create the Task here, but do not start it. 
    listOfTasks.Add(new Task(async() => await blobBlock.UploadFromFileAsync(localFile, FileMode.Create))); 
} 
await Tasks.StartAndWaitAllThrottledAsync(listOfTasks, 5);