그룹의 병렬 실행

Question

간단한 예제에서 내 문제를 설명하고 더 가까운 문제를 설명합니다.

상자 1에 [i1, i2, i3, i4, ...]가 있고 항목을 처리 할 수있는 상자 2가 있습니다 (일반적으로 m은 n보다 훨씬 적음). 각 항목에 필요한 시간이 다릅니다. 나는 모든 항목이 진행될 때까지 항상 m 직업 품목을하고 싶다.

훨씬 더 가까운 문제는 파일의 n 개의 문자열 (URL 주소) 목록 1을 가지고 있고 시스템이 m 개의 파일을 동시에 다운로드하도록하려는 경우 (예 : httpclient.getAsync() 메소드를 통해) . m 개의 항목 중 하나의 다운로드가 끝날 때마다 list1의 다른 나머지 항목을 최대한 빨리 대체해야하며 List1 항목이 모두 진행될 때까지이 항목을 계승해야합니다. (n 및 m의 번호는 런타임에 사용자 입력에 의해 지정됩니다)

어떻게 수행 할 수 있습니까? 동시 작업의 수를 제한 병렬

Answer 1

프로세스 항목 :

string[] strings = GetStrings(); // Items to process. 
const int m = 2; // Max simultaneous jobs. 

Parallel.ForEach(strings, new ParallelOptions {MaxDegreeOfParallelism = m}, s => 
{ 
    DoWork(s); 
}); 

Answer 2

당신은 간단

로

private static HttpClient _client = new HttpClient(); 
public async Task<List<MyClass>> ProcessDownloads(IEnumerable<string> uris, 
                int concurrentDownloads) 
{ 
    var result = new List<MyClass>(); 

    var downloadData = new TransformBlock<string, string>(async uri => 
    { 
     return await _client.GetStringAsync(uri); //GetStringAsync is a thread safe method. 
    }, new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism = concurrentDownloads}); 

    var processData = new TransformBlock<string, MyClass>(
      json => JsonConvert.DeserializeObject<MyClass>(json), 
      new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded}); 

    var collectData = new ActionBlock<MyClass>(
      data => result.Add(data)); //When you don't specifiy options dataflow processes items one at a time. 

    //Set up the chain of blocks, have it call `.Complete()` on the next block when the current block finishes processing it's last item. 
    downloadData.LinkTo(processData, new DataflowLinkOptions {PropagateCompletion = true}); 
    processData.LinkTo(collectData, new DataflowLinkOptions {PropagateCompletion = true}); 

    //Load the data in to the first transform block to start off the process. 
    foreach (var uri in uris) 
    { 
     await downloadData.SendAsync(uri).ConfigureAwait(false); 
    } 
    downloadData.Complete(); //Signal you are done adding data. 

    //Wait for the last object to be added to the list. 
    await collectData.Completion.ConfigureAwait(false); 

    return result; 
} 

입니다 원하는 다음 프로젝트에 System.Threading.Tasks.Dataflow NuGet 패키지를 추가, TPL Dataflow에 보일 것입니다

위의 코드에서 주어진 시간에만 HttpClient가 활성화되고, 무제한 스레드는 수신 된 문자열을 처리하여 객체로 변환하고 단일 스레드는 thos를 사용합니다. e 개체를 만들고 목록에 추가합니다.

업데이트 : 여기 만이 여기에 해당

private static HttpClient _client = new HttpClient(); 
public void ProcessDownloads(IEnumerable<string> uris, int concurrentDownloads) 
{ 
    var downloadData = new ActionBlock<string>(async uri => 
    { 
     var response = await _client.GetAsync(uri); //GetAsync is a thread safe method. 
     //do something with response here. 
    }, new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism = concurrentDownloads}); 


    foreach (var uri in uris) 
    { 
     downloadData.Post(uri); 
    } 
    downloadData.Complete(); 

    downloadData.Completion.Wait(); 
} 

Answer 3

요청 사용할 수있는 일반적인 방법이 무엇인지한다는 것을 단순화 된 예입니다.

이 TIn을 문자열 (URL 주소)이라고하면 asyncProcessor는 URL 주소를 입력으로 사용하여 작업을 반환하는 비동기 메서드가됩니다.

이 방법으로 사용되는 SlimSemaphore는 다른 요청이 완료되는 즉시 n 개의 동시 비동기 I/O 요청을 실시간으로 허용합니다. 슬라이딩 윈도우 패턴과 같은 것.

public static Task ForEachAsync<TIn>(
      IEnumerable<TIn> inputEnumerable, 
      Func<TIn, Task> asyncProcessor, 
      int? maxDegreeOfParallelism = null) 
     { 
      int maxAsyncThreadCount = maxDegreeOfParallelism ?? DefaultMaxDegreeOfParallelism; 
      SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount); 

      IEnumerable<Task> tasks = inputEnumerable.Select(async input => 
      { 
       await throttler.WaitAsync().ConfigureAwait(false); 
       try 
       { 
        await asyncProcessor(input).ConfigureAwait(false); 
       } 
       finally 
       { 
        throttler.Release(); 
       } 
      }); 

      return Task.WhenAll(tasks); 
     } 

Answer 4

간단한 조절 솔루션은 SemaphoreSlim입니다. 그들은

var client = new HttpClient(); 
SemaphoreSlim semaphore = new SemaphoreSlim(m, m); //set the max here 
var tasks = new List<Task>(); 

foreach(var url in urls) 
{ 
    // moving the wait here throttles the foreach loop 
    await semaphore.WaitAsync(); 
    tasks.Add(((Func<Task>)(async() => 
    { 
     //await semaphore.WaitAsync(); 
     var response = await client.GetAsync(url); // possibly ConfigureAwait(false) here 
     // do something with response 
     semaphore.Release(); 
    }))()); 
} 

await Task.WhenAll(tasks); 

필요 이것은 그의 DoWork는 비동기입니다 그것을

var client = new HttpClient(); 
var tasks = new HashSet<Task>(); 

foreach(var url in urls) 
{ 
    if(tasks.Count == m) 
    { 
     tasks.Remove(await Task.WhenAny(tasks));    
    } 

    tasks.Add(((Func<Task>)(async() => 
    { 
     var response = await client.GetAsync(url); // possibly ConfigureAwait(false) here 
     // do something with response    
    }))()); 
} 

await Task.WhenAll(tasks);