Wie ordne ich Aufgaben in C#

Question

richtig in Warteschlange Ich habe eine Aufzählung von Elementen (RunData.Demand), die jeweils einige Arbeit darstellen Aufruf einer API über HTTP. Es funktioniert gut, wenn ich nur foreach über alles und rufen Sie die API während jeder Iteration. Allerdings dauert jede Iteration ein oder zwei Sekunden, also würde ich gerne 2-3 Threads ausführen und die Arbeit zwischen ihnen aufteilen. Hier ist, was ich tue:

ThreadPool.SetMaxThreads(2, 5); // Trying to limit the amount of threads 
var tasks = RunData.Demand 
    .Select(service => Task.Run(async delegate 
    { 
     var availabilityResponse = await client.QueryAvailability(service); 
     // Do some other stuff, not really important 
    })); 

await Task.WhenAll(tasks); 

Der client.QueryAvailability Aufruf ruft im Grunde eine API mit der HttpClient Klasse:

public async Task<QueryAvailabilityResponse> QueryAvailability(QueryAvailabilityMultidayRequest request) 
{ 
    var response = await client.PostAsJsonAsync("api/queryavailabilitymultiday", request); 

    if (response.IsSuccessStatusCode) 
    { 
     return await response.Content.ReadAsAsync<QueryAvailabilityResponse>(); 
    } 

    throw new HttpException((int) response.StatusCode, response.ReasonPhrase); 
} 

Dies funktioniert gut für eine Weile, aber schließlich die Dinge beginnen Timing aus. Wenn ich das HttpClient Timeout auf eine Stunde setze, bekomme ich seltsame interne Serverfehler.

Was ich begann zu tun war eine Stoppuhr innerhalb der QueryAvailability Methode, um zu sehen, was los war.

Was passiert, ist alle 1200 Elemente in RunData.Demand werden auf einmal erstellt und alle 1200 await client.PostAsJsonAsync Methoden aufgerufen werden. Es scheint, dass es dann die 2 Threads verwendet, um die Aufgaben langsam zu überprüfen, so dass ich zum Ende Aufgaben habe, die auf 9 oder 10 Minuten gewartet haben.

Hier ist das Verhalten Ich mag würde:

Ich mag würde die 1200 Aufgaben erstellen und sie dann 3-4 zu einem Zeitpunkt ausgeführt als Threads verfügbar sind. Ich mache nicht wollen 1.200 HTTP-Anrufe sofort anstehen.

Gibt es einen guten Weg, dies zu tun?

Answer 1

Wie ich immer empfehlen .. was Sie brauchen, ist TPL Dataflow (zu installieren: Install-Package Microsoft.Tpl.Dataflow).

Sie erstellen eine ActionBlock mit einer Aktion für jedes Element auszuführen. Stellen Sie MaxDegreeOfParallelism für die Drosselung ein. Starten Sie in sie zu veröffentlichen und warten auf ihre Fertigstellung:

var block = new ActionBlock<QueryAvailabilityMultidayRequest>(async service => 
{ 
    var availabilityResponse = await client.QueryAvailability(service); 
    // ... 
}, 
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }); 

foreach (var service in RunData.Demand) 
{ 
    block.Post(service); 
} 

block.Complete(); 
await block.Completion; 

Answer 2

Sie sind Asynchron-HTTP-Aufrufe verwenden, so die Anzahl der Threads Begrenzung nicht helfen (noch wird ParallelOptions.MaxDegreeOfParallelism in Parallel.ForEach als eine der Antworten schon sagt). Selbst ein einzelner Thread kann alle Anfragen initiieren und die Ergebnisse so verarbeiten, wie sie ankommen.

Eine Möglichkeit, es zu lösen, ist die Verwendung von TPL Dataflow.

Eine weitere schöne Lösung ist die Quelle IEnumerable in Partitionen und Prozesselemente in jeder Partition nacheinander zu unterteilen, wie in this blog post beschrieben:

public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll(
     from partition in Partitioner.Create(source).GetPartitions(dop) 
     select Task.Run(async delegate 
     { 
      using (partition) 
       while (partition.MoveNext()) 
        await body(partition.Current); 
     })); 
} 

Answer 3

Während die Datenfluß-Bibliothek ist sehr gut, ich denke, es ist ein bisschen schwer ist, wenn sie nicht mit Blockzusammensetzung. Ich würde dazu neigen, etwas wie die Verlängerungsmethode unten zu verwenden. Im Gegensatz zur Partitioner-Methode führt dies die asynchronen Methoden im aufrufenden Kontext aus. Der Cloat ist, dass, wenn Ihr Code nicht wirklich asynchron ist oder einen "schnellen Pfad" verwendet, er synchron läuft, da keine Threads vorhanden sind werden explizit erstellt.

public static async Task RunParallelAsync<T>(this IEnumerable<T> items, Func<T, Task> asyncAction, int maxParallel) 
{ 
    var tasks = new List<Task>(); 

    foreach (var item in items) 
    { 
     tasks.Add(asyncAction(item)); 

     if (tasks.Count < maxParallel) 
       continue; 

     var notCompleted = tasks.Where(t => !t.IsCompleted).ToList(); 

     if (notCompleted.Count >= maxParallel) 
      await Task.WhenAny(notCompleted); 
    } 

    await Task.WhenAll(tasks); 
} 

Answer 4

alte Frage, aber ich möchte eine alternative leichte Lösung mit der SemaphoreSlim Klasse vorzuschlagen. Referenzieren Sie einfach System.Threading.

SemaphoreSlim sem = new SemaphoreSlim(4,4); 

foreach (var service in RunData.Demand) 
{ 

    await sem.WaitAsync(); 
    Task t = Task.Run(async() => 
    { 
     var availabilityResponse = await client.QueryAvailability(serviceCopy));  
     // do your other stuff here with the result of QueryAvailability 
    } 
    t.ContinueWith(sem.Release()); 
} 

Der Semaphor fungiert als Verriegelungsmechanismus. Sie können das Semaphor nur aufrufen, indem Sie Wait (WaitAsync) aufrufen, das eins von der Zählung subtrahiert. Anruferklärung fügt eins zur Zählung hinzu.