Parallele Ausführung von Aufgaben in Gruppen

Question

Ich beschreibe mein Problem in einem einfachen Beispiel und beschreibe dann ein engeres Problem.

Stellen Sie sich vor Wir haben n Elemente [i1, i2, i3, i4, ..., in] in der Box1 und wir haben eine Box2, die m Elemente verarbeiten kann (m ist normalerweise viel kleiner als n). Die für jedes Element erforderliche Zeit ist unterschiedlich. Ich möchte immer m Job Items machen, bis alle Items bearbeitet sind.

Ein viel engeres Problem ist, dass Sie zum Beispiel eine Liste1 von n Zeichenfolgen (URL-Adressen) von Dateien haben und wir ein System haben wollen, um m Dateien gleichzeitig herunterzuladen (zum Beispiel via httpclient.getAsync() Methode) . Immer wenn das Herunterladen eines der m Elemente beendet wird, muss ein anderes verbleibendes Element von list1 so schnell wie möglich ersetzt werden. Dies muss so lange fortgesetzt werden, bis alle Elemente von List1 ausgeführt wurden. (Anzahl der n und m werden von Benutzern zur Laufzeit angegeben)

Wie kann dies getan werden?

Answer 1

Prozesselemente parallel die Anzahl der gleichzeitig Arbeitsplätze zu beschränken:

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

Sie sollten die System.Threading.Tasks.Dataflow NuGet Paket zu einem Projekt suchen Sie dann in zu TPL Dataflow, fügen Sie, was Sie wollen so einfach wie

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

In dem obigen Code nur concurrentDownloads Anzahl von HttpClients wird zu jeder beliebigen Zeit aktiv sein, unbegrenzte Threads werden die empfangenen Strings verarbeiten und sie in Objekte konvertieren, und ein einzelner Thread wird thos nehmen Objekte und fügen Sie sie einer Liste hinzu.

UPDATE: hier ist ein vereinfachtes Beispiel, dass nur das tut, was sie in der Frage gestellt ist

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

Hier ist eine generische Methode, die Sie verwenden können.

Wenn Sie dies aufrufen, wird TIn eine Zeichenfolge (URL-Adressen) sein und der asyncProcessor wird Ihre asynchrone Methode sein, die die URL-Adresse als Eingabe verwendet und eine Aufgabe zurückgibt.

Der von dieser Methode verwendete SlimSemaphore wird in Echtzeit nur eine Anzahl von gleichzeitigen asynchronen E/A-Anforderungen zulassen, sobald die andere Anforderung ausgeführt wird. So etwas wie ein gleitendes Fenstermuster.

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

Eine einfache Lösung für die Drosselung ist ein SemaphoreSlim.
EDIT
Nach einer leichten Änderung erstellt der Code nun die Aufgaben, wenn sie benötigt werden

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

Dies ist ein weiterer Weg, es zu tun

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