.NET Framework
Task Parallel Library (TPL)
Szukaj…
Uwagi
Cele i przypadki użycia
Celem biblioteki zadań równoległych jest uproszczenie procesu pisania i utrzymywania wielowątkowego i równoległego kodu.
Niektóre przypadki użycia *:
- Utrzymanie responsywności interfejsu użytkownika poprzez uruchomienie pracy w tle dla oddzielnego zadania
- Podział obciążenia
- Zezwolenie aplikacji klienckiej na jednoczesne wysyłanie i odbieranie żądań (reszta, TCP / UDP, ect)
- Odczytywanie i / lub pisanie wielu plików jednocześnie
* Kod powinien być rozpatrywany indywidualnie dla wielowątkowości. Na przykład, jeśli pętla ma tylko kilka iteracji lub wykonuje jedynie niewielką część pracy, narzut związany z równoległością może przeważyć korzyści.
TPL z .Net 3.5
TPL jest również dostępna dla .Net 3.5 zawartego w pakiecie NuGet, nazywa się Task Parallel Library.
Podstawowa pętla producent-konsument (BlockingCollection)
var collection = new BlockingCollection<int>(5);
var random = new Random();
var producerTask = Task.Run(() => {
for(int item=1; item<=10; item++)
{
collection.Add(item);
Console.WriteLine("Produced: " + item);
Thread.Sleep(random.Next(10,1000));
}
collection.CompleteAdding();
Console.WriteLine("Producer completed!");
});
Warto zauważyć, że jeśli nie wywołasz collection.CompleteAdding();
, możesz dodawać do kolekcji, nawet jeśli Twoje zadanie konsumenckie jest uruchomione. Wystarczy wywołać collection.CompleteAdding();
kiedy jesteś pewien, że nie ma już żadnych dodatków. Funkcji tej można użyć, aby utworzyć wzór dla wielu producentów na jednego konsumenta, w którym istnieje wiele źródeł, które podają produkty do BlockingCollection, a jeden konsument wyciąga produkty i coś z nimi robi. Jeśli twoja BlockingCollection jest pusta przed wywołaniem pełnego dodawania, Enumerable from collection.GetConsumingEnumerable()
będzie blokować do momentu dodania nowego elementu do kolekcji lub BlockingCollection.CompleteAdding (); jest wywoływane, a kolejka jest pusta.
var consumerTask = Task.Run(() => {
foreach(var item in collection.GetConsumingEnumerable())
{
Console.WriteLine("Consumed: " + item);
Thread.Sleep(random.Next(10,1000));
}
Console.WriteLine("Consumer completed!");
});
Task.WaitAll(producerTask, consumerTask);
Console.WriteLine("Everything completed!");
Zadanie: podstawowa instancja i oczekiwanie
Zadanie można utworzyć bezpośrednio tworząc instancję klasy Task
...
var task = new Task(() =>
{
Console.WriteLine("Task code starting...");
Thread.Sleep(2000);
Console.WriteLine("...task code ending!");
});
Console.WriteLine("Starting task...");
task.Start();
task.Wait();
Console.WriteLine("Task completed!");
... lub za pomocą statycznej metody Task.Run
:
Console.WriteLine("Starting task...");
var task = Task.Run(() =>
{
Console.WriteLine("Task code starting...");
Thread.Sleep(2000);
Console.WriteLine("...task code ending!");
});
task.Wait();
Console.WriteLine("Task completed!");
Pamiętaj, że tylko w pierwszym przypadku konieczne jest jawne wywołanie Start
.
Zadanie: WaitAll i przechwytywanie zmiennych
var tasks = Enumerable.Range(1, 5).Select(n => new Task<int>(() =>
{
Console.WriteLine("I'm task " + n);
return n;
})).ToArray();
foreach(var task in tasks) task.Start();
Task.WaitAll(tasks);
foreach(var task in tasks)
Console.WriteLine(task.Result);
Zadanie: WaitAny
var allTasks = Enumerable.Range(1, 5).Select(n => new Task<int>(() => n)).ToArray();
var pendingTasks = allTasks.ToArray();
foreach(var task in allTasks) task.Start();
while(pendingTasks.Length > 0)
{
var finishedTask = pendingTasks[Task.WaitAny(pendingTasks)];
Console.WriteLine("Task {0} finished", finishedTask.Result);
pendingTasks = pendingTasks.Except(new[] {finishedTask}).ToArray();
}
Task.WaitAll(allTasks);
Uwaga: Ostateczne WaitAll
jest konieczne, ponieważ WaitAny
nie powoduje wyjątków.
Zadanie: obsługa wyjątków (za pomocą funkcji Wait)
var task1 = Task.Run(() =>
{
Console.WriteLine("Task 1 code starting...");
throw new Exception("Oh no, exception from task 1!!");
});
var task2 = Task.Run(() =>
{
Console.WriteLine("Task 2 code starting...");
throw new Exception("Oh no, exception from task 2!!");
});
Console.WriteLine("Starting tasks...");
try
{
Task.WaitAll(task1, task2);
}
catch(AggregateException ex)
{
Console.WriteLine("Task(s) failed!");
foreach(var inner in ex.InnerExceptions)
Console.WriteLine(inner.Message);
}
Console.WriteLine("Task 1 status is: " + task1.Status); //Faulted
Console.WriteLine("Task 2 status is: " + task2.Status); //Faulted
Zadanie: obsługa wyjątków (bez użycia funkcji czekania)
var task1 = Task.Run(() =>
{
Console.WriteLine("Task 1 code starting...");
throw new Exception("Oh no, exception from task 1!!");
});
var task2 = Task.Run(() =>
{
Console.WriteLine("Task 2 code starting...");
throw new Exception("Oh no, exception from task 2!!");
});
var tasks = new[] {task1, task2};
Console.WriteLine("Starting tasks...");
while(tasks.All(task => !task.IsCompleted));
foreach(var task in tasks)
{
if(task.IsFaulted)
Console.WriteLine("Task failed: " +
task.Exception.InnerExceptions.First().Message);
}
Console.WriteLine("Task 1 status is: " + task1.Status); //Faulted
Console.WriteLine("Task 2 status is: " + task2.Status); //Faulted
Zadanie: anulowanie za pomocą CancellationToken
var cancellationTokenSource = new CancellationTokenSource();
var cancellationToken = cancellationTokenSource.Token;
var task = new Task((state) =>
{
int i = 1;
var myCancellationToken = (CancellationToken)state;
while(true)
{
Console.Write("{0} ", i++);
Thread.Sleep(1000);
myCancellationToken.ThrowIfCancellationRequested();
}
},
cancellationToken: cancellationToken,
state: cancellationToken);
Console.WriteLine("Counting to infinity. Press any key to cancel!");
task.Start();
Console.ReadKey();
cancellationTokenSource.Cancel();
try
{
task.Wait();
}
catch(AggregateException ex)
{
ex.Handle(inner => inner is OperationCanceledException);
}
Console.WriteLine($"{Environment.NewLine}You have cancelled! Task status is: {task.Status}");
//Canceled
Alternatywnie do ThrowIfCancellationRequested
żądanie anulowania można wykryć za pomocą IsCancellationRequested
a IsCancellationRequested
OperationCanceledException
można zgłosić ręcznie:
//New task delegate
int i = 1;
var myCancellationToken = (CancellationToken)state;
while(!myCancellationToken.IsCancellationRequested)
{
Console.Write("{0} ", i++);
Thread.Sleep(1000);
}
Console.WriteLine($"{Environment.NewLine}Ouch, I have been cancelled!!");
throw new OperationCanceledException(myCancellationToken);
Zwróć uwagę, w jaki sposób token anulowania jest przekazywany do konstruktora zadania w parametrze cancellationToken
. Jest to konieczne w celu umożliwienia przejścia zadaniowe do Canceled
państwa, a nie do Faulted
stanie, gdy ThrowIfCancellationRequested
jest wywoływany. Z tego samego powodu token anulowania jest jawnie podany w konstruktorze OperationCanceledException
w drugim przypadku.
Task.WhenAny
var random = new Random();
IEnumerable<Task<int>> tasks = Enumerable.Range(1, 5).Select(n => Task.Run(async() =>
{
Console.WriteLine("I'm task " + n);
await Task.Delay(random.Next(10,1000));
return n;
}));
Task<Task<int>> whenAnyTask = Task.WhenAny(tasks);
Task<int> completedTask = await whenAnyTask;
Console.WriteLine("The winner is: task " + await completedTask);
await Task.WhenAll(tasks);
Console.WriteLine("All tasks finished!");
Task.WhenAll
var random = new Random();
IEnumerable<Task<int>> tasks = Enumerable.Range(1, 5).Select(n => Task.Run(() =>
{
Console.WriteLine("I'm task " + n);
return n;
}));
Task<int[]> task = Task.WhenAll(tasks);
int[] results = await task;
Console.WriteLine(string.Join(",", results.Select(n => n.ToString())));
// Output: 1,2,3,4,5
Parallel.Invoke
var actions = Enumerable.Range(1, 10).Select(n => new Action(() =>
{
Console.WriteLine("I'm task " + n);
if((n & 1) == 0)
throw new Exception("Exception from task " + n);
})).ToArray();
try
{
Parallel.Invoke(actions);
}
catch(AggregateException ex)
{
foreach(var inner in ex.InnerExceptions)
Console.WriteLine("Task failed: " + inner.Message);
}
Parallel.ForEach
W tym przykładzie użyto Parallel.ForEach
do obliczenia sumy liczb od 1 do 10000 przy użyciu wielu wątków. Aby uzyskać bezpieczeństwo wątków, do sumowania liczb służy Interlocked.Add
.
using System.Threading;
int Foo()
{
int total = 0;
var numbers = Enumerable.Range(1, 10000).ToList();
Parallel.ForEach(numbers,
() => 0, // initial value,
(num, state, localSum) => num + localSum,
localSum => Interlocked.Add(ref total, localSum));
return total; // total = 50005000
}
Równoległy
W tym przykładzie użyto Parallel.For
do obliczenia sumy liczb od 1 do 10000 przy użyciu wielu wątków. Aby uzyskać bezpieczeństwo wątków, do sumowania liczb służy Interlocked.Add
.
using System.Threading;
int Foo()
{
int total = 0;
Parallel.For(1, 10001,
() => 0, // initial value,
(num, state, localSum) => num + localSum,
localSum => Interlocked.Add(ref total, localSum));
return total; // total = 50005000
}
Płynny kontekst wykonania z AsyncLocal
Gdy musisz przekazać niektóre dane z zadania nadrzędnego do jego zadań AsyncLocal
, aby logicznie przepływało ono wraz z wykonaniem, użyj klasy AsyncLocal
:
void Main()
{
AsyncLocal<string> user = new AsyncLocal<string>();
user.Value = "initial user";
// this does not affect other tasks - values are local relative to the branches of execution flow
Task.Run(() => user.Value = "user from another task");
var task1 = Task.Run(() =>
{
Console.WriteLine(user.Value); // outputs "initial user"
Task.Run(() =>
{
// outputs "initial user" - value has flown from main method to this task without being changed
Console.WriteLine(user.Value);
}).Wait();
user.Value = "user from task1";
Task.Run(() =>
{
// outputs "user from task1" - value has flown from main method to task1
// than value was changed and flown to this task.
Console.WriteLine(user.Value);
}).Wait();
});
task1.Wait();
// ouputs "initial user" - changes do not propagate back upstream the execution flow
Console.WriteLine(user.Value);
}
Uwaga: Jak widać z powyższego przykładu, AsynLocal.Value
ma copy on read
semantycznym copy on read
, ale jeśli przepłyniesz jakiś typ odwołania i zmienisz jego właściwości, AsynLocal.Value
na inne zadania. Dlatego najlepszą praktyką w AsyncLocal
jest używanie typów wartości lub typów niezmiennych.
Parallel.ForEach w VB.NET
For Each row As DataRow In FooDataTable.Rows
Me.RowsToProcess.Add(row)
Next
Dim myOptions As ParallelOptions = New ParallelOptions()
myOptions.MaxDegreeOfParallelism = environment.processorcount
Parallel.ForEach(RowsToProcess, myOptions, Sub(currentRow, state)
ProcessRowParallel(currentRow, state)
End Sub)
Zadanie: zwracanie wartości
Zadanie, które zwraca wartość, ma typ zwrotu Task< TResult >
gdzie TResult jest rodzajem wartości, którą należy zwrócić. Możesz zapytać o wynik zadania według jego właściwości Result.
Task<int> t = Task.Run(() =>
{
int sum = 0;
for(int i = 0; i < 500; i++)
sum += i;
return sum;
});
Console.WriteLine(t.Result); // Outuput 124750
Jeśli zadanie zostanie wykonane asynchronicznie, a oczekiwanie na zadanie zwróci wynik, wynik.
public async Task DoSomeWork()
{
WebClient client = new WebClient();
// Because the task is awaited, result of the task is assigned to response
string response = await client.DownloadStringTaskAsync("http://somedomain.com");
}