Procedure: De mate van parallelle uitvoering opgeven in een gegevensstroomblok
In dit document wordt beschreven hoe u de ExecutionDataflowBlockOptions.MaxDegreeOfParallelism eigenschap instelt om een gegevensstroomblok voor uitvoering in te schakelen voor het verwerken van meer dan één bericht tegelijk. Dit is handig wanneer u een gegevensstroomblok hebt dat een langlopende berekening uitvoert en kan profiteren van het parallel verwerken van berichten. In dit voorbeeld wordt de System.Threading.Tasks.Dataflow.ActionBlock<TInput> klasse gebruikt om gelijktijdig meerdere gegevensstroombewerkingen uit te voeren. U kunt echter de maximale mate van parallelle uitvoering opgeven in een van de vooraf gedefinieerde uitvoeringsbloktypen die de TPL-gegevensstroombibliotheek biedt, ActionBlock<TInput>, en System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput>System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput>.
Notitie
De TPL-gegevensstroombibliotheek (de System.Threading.Tasks.Dataflow naamruimte) wordt niet gedistribueerd met .NET. Als u de System.Threading.Tasks.Dataflow naamruimte in Visual Studio wilt installeren, opent u uw project, kiest u NuGet-pakketten beheren in het menu Project en zoekt u online naar het System.Threading.Tasks.Dataflow pakket. U kunt het ook installeren met behulp van de .NET Core CLI.dotnet add package System.Threading.Tasks.Dataflow
Opmerking
In het volgende voorbeeld worden twee berekeningen voor gegevensstromen uitgevoerd en wordt de verstreken tijd afgedrukt die vereist is voor elke berekening. De eerste berekening geeft een maximale mate van parallelle uitvoering van 1 aan, wat de standaardwaarde is. Een maximale mate van parallelle uitvoering van 1 zorgt ervoor dat het gegevensstroomblok berichten serieel verwerkt. De tweede berekening lijkt op de eerste, behalve dat hiermee een maximale mate van parallelle uitvoering wordt opgegeven die gelijk is aan het aantal beschikbare processors. Hierdoor kan het gegevensstroomblok meerdere bewerkingen parallel uitvoeren.
using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks.Dataflow;
// Demonstrates how to specify the maximum degree of parallelism
// when using dataflow.
class Program
{
// Performs several computations by using dataflow and returns the elapsed
// time required to perform the computations.
static TimeSpan TimeDataflowComputations(int maxDegreeOfParallelism,
int messageCount)
{
// Create an ActionBlock<int> that performs some work.
var workerBlock = new ActionBlock<int>(
// Simulate work by suspending the current thread.
millisecondsTimeout => Thread.Sleep(millisecondsTimeout),
// Specify a maximum degree of parallelism.
new ExecutionDataflowBlockOptions
{
MaxDegreeOfParallelism = maxDegreeOfParallelism
});
// Compute the time that it takes for several messages to
// flow through the dataflow block.
Stopwatch stopwatch = new Stopwatch();
stopwatch.Start();
for (int i = 0; i < messageCount; i++)
{
workerBlock.Post(1000);
}
workerBlock.Complete();
// Wait for all messages to propagate through the network.
workerBlock.Completion.Wait();
// Stop the timer and return the elapsed number of milliseconds.
stopwatch.Stop();
return stopwatch.Elapsed;
}
static void Main(string[] args)
{
int processorCount = Environment.ProcessorCount;
int messageCount = processorCount;
// Print the number of processors on this computer.
Console.WriteLine("Processor count = {0}.", processorCount);
TimeSpan elapsed;
// Perform two dataflow computations and print the elapsed
// time required for each.
// This call specifies a maximum degree of parallelism of 1.
// This causes the dataflow block to process messages serially.
elapsed = TimeDataflowComputations(1, messageCount);
Console.WriteLine("Degree of parallelism = {0}; message count = {1}; " +
"elapsed time = {2}ms.", 1, messageCount, (int)elapsed.TotalMilliseconds);
// Perform the computations again. This time, specify the number of
// processors as the maximum degree of parallelism. This causes
// multiple messages to be processed in parallel.
elapsed = TimeDataflowComputations(processorCount, messageCount);
Console.WriteLine("Degree of parallelism = {0}; message count = {1}; " +
"elapsed time = {2}ms.", processorCount, messageCount, (int)elapsed.TotalMilliseconds);
}
}
/* Sample output:
Processor count = 4.
Degree of parallelism = 1; message count = 4; elapsed time = 4032ms.
Degree of parallelism = 4; message count = 4; elapsed time = 1001ms.
*/
Imports System.Diagnostics
Imports System.Threading
Imports System.Threading.Tasks.Dataflow
' Demonstrates how to specify the maximum degree of parallelism
' when using dataflow.
Friend Class Program
' Performs several computations by using dataflow and returns the elapsed
' time required to perform the computations.
Private Shared Function TimeDataflowComputations(ByVal maxDegreeOfParallelism As Integer, ByVal messageCount As Integer) As TimeSpan
' Create an ActionBlock<int> that performs some work.
Dim workerBlock = New ActionBlock(Of Integer)(Function(millisecondsTimeout) Pause(millisecondsTimeout), New ExecutionDataflowBlockOptions() With {.MaxDegreeOfParallelism = maxDegreeOfParallelism})
' Simulate work by suspending the current thread.
' Specify a maximum degree of parallelism.
' Compute the time that it takes for several messages to
' flow through the dataflow block.
Dim stopwatch As New Stopwatch()
stopwatch.Start()
For i As Integer = 0 To messageCount - 1
workerBlock.Post(1000)
Next i
workerBlock.Complete()
' Wait for all messages to propagate through the network.
workerBlock.Completion.Wait()
' Stop the timer and return the elapsed number of milliseconds.
stopwatch.Stop()
Return stopwatch.Elapsed
End Function
Private Shared Function Pause(ByVal obj As Object)
Thread.Sleep(obj)
Return Nothing
End Function
Shared Sub Main(ByVal args() As String)
Dim processorCount As Integer = Environment.ProcessorCount
Dim messageCount As Integer = processorCount
' Print the number of processors on this computer.
Console.WriteLine("Processor count = {0}.", processorCount)
Dim elapsed As TimeSpan
' Perform two dataflow computations and print the elapsed
' time required for each.
' This call specifies a maximum degree of parallelism of 1.
' This causes the dataflow block to process messages serially.
elapsed = TimeDataflowComputations(1, messageCount)
Console.WriteLine("Degree of parallelism = {0}; message count = {1}; " & "elapsed time = {2}ms.", 1, messageCount, CInt(Fix(elapsed.TotalMilliseconds)))
' Perform the computations again. This time, specify the number of
' processors as the maximum degree of parallelism. This causes
' multiple messages to be processed in parallel.
elapsed = TimeDataflowComputations(processorCount, messageCount)
Console.WriteLine("Degree of parallelism = {0}; message count = {1}; " & "elapsed time = {2}ms.", processorCount, messageCount, CInt(Fix(elapsed.TotalMilliseconds)))
End Sub
End Class
' Sample output:
'Processor count = 4.
'Degree of parallelism = 1; message count = 4; elapsed time = 4032ms.
'Degree of parallelism = 4; message count = 4; elapsed time = 1001ms.
'
Robuuste programmering
Standaard worden berichten door elk vooraf gedefinieerd gegevensstroomblok doorgegeven in de volgorde waarin de berichten worden ontvangen. Hoewel meerdere berichten tegelijkertijd worden verwerkt wanneer u een maximale mate van parallelle uitvoering opgeeft die groter is dan 1, worden ze nog steeds doorgegeven in de volgorde waarin ze worden ontvangen.
Omdat de MaxDegreeOfParallelism eigenschap de maximale mate van parallelle uitvoering aangeeft, kan het gegevensstroomblok worden uitgevoerd met een mindere mate van parallelle uitvoering dan u opgeeft. Het gegevensstroomblok kan minder parallellisme gebruiken om te voldoen aan de functionele vereisten of om rekening te houden met een gebrek aan beschikbare systeembronnen. Een gegevensstroomblok kiest nooit een grotere mate van parallelle uitvoering dan u opgeeft.
