Jak prawidłowo ustawiać kolejki zadań do uruchomienia w języku C#

Question

Mam wyliczenie elementów (RunData.Demand), z których każda przedstawia niektóre prace związane z wywoływaniem API przez HTTP. Działa to świetnie, jeśli tylko przez cały czas będę je wywoływać i wywoływać API podczas każdej iteracji. Jednak każda iteracja trwa sekundę lub dwie, więc chciałbym uruchomić 2-3 wątki i podzielić pracę między nimi. Oto, co robię:

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

client.QueryAvailability wezwanie zasadzie wywołuje API przy użyciu klasy HttpClient:

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

Działa to doskonale na chwilę, ale w końcu wszystko zaczyna odmierzanie. Jeśli ustawię limit czasu HttpClient na godzinę, wtedy zaczną się dziwne błędy wewnętrznego serwera.

Zacząłem od ustawienia stopera w metodzie QueryAvailability, aby sprawdzić, co się dzieje.

Co się dzieje, to 1200 pozycji w RunData.Demand są tworzone jednocześnie i wszystkie metody 1200 await client.PostAsJsonAsync są wywoływane. Wygląda na to, że używa 2 wątków do powolnego sprawdzania zadań, więc pod koniec mam zadania, które czekają na 9 lub 10 minut.

Oto zachowanie Chciałbym:

Chciałbym stworzyć 1200 zadań, a następnie uruchomić je 3-4 naraz stają się dostępne wątki. Robię , a nie chcę od razu kolejkować 1 200 połączeń HTTP.

Czy jest dobry sposób na robienie tego?

Answer 1

Jak zawsze polecam .. to czego potrzebujesz to TPL Dataflow (do instalacji: Install-Package Microsoft.Tpl.Dataflow).

Tworzysz ActionBlock z akcją do wykonania na każdym przedmiocie. Ustaw MaxDegreeOfParallelism dla dławienia. Rozpocząć delegowania do niego i czeka na jego zakończenie:

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

Używasz połączeń asynchronicznych HTTP, więc ograniczenie liczby wątków nie pomoże (nie będzie ParallelOptions.MaxDegreeOfParallelism w Parallel.ForEach jako jedna z odpowiedzi wskazuje). Nawet pojedynczy wątek może inicjować wszystkie żądania i przetwarzać wyniki po ich otrzymaniu.

Jednym ze sposobów rozwiązania tego problemu jest wykorzystanie przepływu danych TPL.

Innym miłym rozwiązaniem jest podzielenie źródło IEnumerable do przegród i elementów procesowych w każdym sekwencyjnie działowej, jak opisano w this blog post:

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

Choć Dataflow biblioteka jest super, myślę, że to trochę ciężkie, gdy nie jest używany skład bloku. Chciałbym użyć czegoś podobnego do metody rozszerzenia poniżej.

Również, w przeciwieństwie do metody Partitioner, uruchamia metody asynchroniczne w kontekście wywołującym - co jest zastrzeżeniem, że jeśli twój kod nie jest prawdziwie asynchroniczny lub przyjmuje "szybką ścieżkę", to będzie skutecznie działał synchronicznie, ponieważ nie ma wątków są jawnie tworzone.

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

Stare pytanie, ale chciałbym zaproponować alternatywną lekkie rozwiązanie przy użyciu klasy SemaphoreSlim. Wystarczy odwołać się do 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()); 
} 

Semafor działa jak mechanizm blokujący. Możesz wejść do semafora tylko przez wywołanie Czekaj (WaitAsync), które odejmuje jeden od licznika. Wywołanie wywołania dodaje jeden do liczby.