Dataflöde (parallellt aktivitetsbibliotek)
Det parallella aktivitetsbiblioteket (TPL) innehåller dataflödeskomponenter för att öka robustheten i samtidighetsaktiverade program. Dessa dataflödeskomponenter kallas gemensamt för TPL-dataflödesbiblioteket. Den här dataflödesmodellen främjar aktörsbaserad programmering genom att tillhandahålla processmeddelandeöverföring för grovkorniga dataflöden och pipelining-uppgifter. Dataflödeskomponenterna bygger på typerna och schemaläggningsinfrastrukturen för TPL och integreras med språkstödet C#, Visual Basic och F# för asynkron programmering. Dessa dataflödeskomponenter är användbara när du har flera åtgärder som måste kommunicera med varandra asynkront eller när du vill bearbeta data när de blir tillgängliga. Tänk dig till exempel ett program som bearbetar bilddata från en webbkamera. Genom att använda dataflödesmodellen kan programmet bearbeta bildramar när de blir tillgängliga. Om programmet förbättrar bildramarna, till exempel genom att utföra ljuskorrigering eller minskning av röda ögon, kan du skapa en pipeline med dataflödeskomponenter. Varje steg i pipelinen kan använda mer grova parallellitetsfunktioner, till exempel de funktioner som tillhandahålls av TPL, för att transformera avbildningen.
Det här dokumentet innehåller en översikt över TPL-dataflödesbiblioteket. Den beskriver programmeringsmodellen, de fördefinierade blocktyperna för dataflöden och hur du konfigurerar dataflödesblock för att uppfylla de specifika kraven för dina program.
Kommentar
TPL-dataflödesbiblioteket ( System.Threading.Tasks.Dataflow namnområdet) distribueras inte med .NET. Om du vill installera System.Threading.Tasks.Dataflow namnområdet i Visual Studio öppnar du projektet, väljer Hantera NuGet-paket på Project-menyn och söker online efter System.Threading.Tasks.Dataflow paketet. Du kan också installera den med .NET Core CLI genom att köra dotnet add package System.Threading.Tasks.Dataflow.
Programmeringsmodell
TPL-dataflödesbiblioteket utgör en grund för meddelandeöverföring och parallellisering av processorintensiva och I/O-intensiva program som har högt dataflöde och låg svarstid. Det ger dig också explicit kontroll över hur data buffrads och flyttas runt i systemet. För att bättre förstå programmeringsmodellen för dataflöden bör du överväga ett program som asynkront läser in avbildningar från disken och skapar en sammansatt av dessa avbildningar. Traditionella programmeringsmodeller kräver vanligtvis att du använder återanrop och synkroniseringsobjekt, till exempel lås, för att samordna uppgifter och åtkomst till delade data. Genom att använda programmeringsmodellen för dataflöde kan du skapa dataflödesobjekt som bearbetar bilder när de läse från disk. Under dataflödesmodellen deklarerar du hur data hanteras när de blir tillgängliga och även eventuella beroenden mellan data. Eftersom körningen hanterar beroenden mellan data kan du ofta undvika kravet på att synkronisera åtkomst till delade data. Eftersom körningsscheman fungerar baserat på den asynkrona ankomsten av data kan dataflödet dessutom förbättra svarstiden och dataflödet genom att effektivt hantera de underliggande trådarna. Ett exempel som använder programmeringsmodellen för dataflöde för att implementera bildbearbetning i ett Windows Forms-program finns i Genomgång: Använda dataflöde i ett Windows Forms-program.
Källor och mål
TPL-dataflödesbiblioteket består av dataflödesblock, som är datastrukturer som buffrar och bearbetar data. TPL definierar tre typer av dataflödesblock: källblock, målblock och spridningsblock. Ett källblock fungerar som en datakälla och kan läsas från. Ett målblock fungerar som mottagare av data och kan skrivas till. Ett spridningsblock fungerar både som ett källblock och ett målblock och kan läsas från och skrivas till. TPL definierar System.Threading.Tasks.Dataflow.ISourceBlock<TOutput> gränssnittet för att representera källor, System.Threading.Tasks.Dataflow.ITargetBlock<TInput> representera mål och System.Threading.Tasks.Dataflow.IPropagatorBlock<TInput,TOutput> representera spridningsfaktorer. IPropagatorBlock<TInput,TOutput> ärver från både ISourceBlock<TOutput>, och ITargetBlock<TInput>.
TPL-dataflödesbiblioteket innehåller flera fördefinierade blocktyper för dataflöden som implementerar gränssnitten ISourceBlock<TOutput>, ITargetBlock<TInput>och IPropagatorBlock<TInput,TOutput> . Dessa typer av dataflödesblock beskrivs i det här dokumentet i avsnittet Fördefinierade typer av dataflödesblock.
Anslut block
Du kan ansluta dataflödesblock för att bilda pipelines, som är linjära sekvenser av dataflödesblock eller nätverk, som är diagram över dataflödesblock. En pipeline är en form av nätverk. I en pipeline eller ett nätverk sprider källor asynkront data till mål när dessa data blir tillgängliga. Metoden ISourceBlock<TOutput>.LinkTo länkar ett källdataflödesblock till ett målblock. En källa kan länkas till noll eller fler mål. mål kan länkas från noll eller flera källor. Du kan lägga till eller ta bort dataflödesblock till eller från en pipeline eller ett nätverk samtidigt. De fördefinierade blocktyperna för dataflöden hanterar alla trådsäkerhetsaspekter för länkning och avlänkning.
Ett exempel som ansluter dataflödesblock för att bilda en grundläggande pipeline finns i Genomgång: Skapa en dataflödespipeline. Ett exempel som ansluter dataflödesblock till ett mer komplext nätverk finns i Genomgång: Använda dataflöde i ett Windows Forms-program. Ett exempel som avlänkar ett mål från en källa när källan erbjuder målet ett meddelande finns i Så här gör du: Ta bort länk till dataflödesblock.
Filtrering
När du anropar ISourceBlock<TOutput>.LinkTo metoden för att länka en källa till ett mål kan du ange ett ombud som avgör om målblocket accepterar eller avvisar ett meddelande baserat på värdet för meddelandet. Den här filtreringsmekanismen är ett användbart sätt att garantera att ett dataflödesblock endast tar emot vissa värden. För de flesta fördefinierade typer av dataflödesblock erbjuder källan meddelandet till nästa mål om ett källblock är anslutet till flera målblock när ett målblock avvisar ett meddelande. I vilken ordning en källa erbjuder meddelanden till mål definieras av källan och kan variera beroende på källans typ. De flesta typer av källblock slutar att erbjuda ett meddelande när ett mål accepterar meddelandet. Ett undantag till den BroadcastBlock<T> här regeln är klassen, som erbjuder varje meddelande till alla mål, även om vissa mål avvisar meddelandet. Ett exempel som använder filtrering för att endast bearbeta vissa meddelanden finns i Genomgång: Använda dataflöde i ett Windows Forms-program.
Viktigt!
Eftersom varje fördefinierad typ av källdataflödesblock garanterar att meddelanden sprids ut i den ordning de tas emot, måste varje meddelande läsas från källblocket innan källblocket kan bearbeta nästa meddelande. När du använder filtrering för att ansluta flera mål till en källa kontrollerar du därför att minst ett målblock tar emot varje meddelande. Annars kan programmet blockeras.
Meddelandeöverföring
Programmeringsmodellen för dataflöde är relaterad till begreppet meddelandeöverföring, där oberoende komponenter i ett program kommunicerar med varandra genom att skicka meddelanden. Ett sätt att sprida meddelanden mellan programkomponenter är att anropa Post (synkrona) och SendAsync (asynkrona) metoder för att skicka meddelanden till måldataflödesblock och Receivemetoderna , ReceiveAsyncoch TryReceive för att ta emot meddelanden från källblock. Du kan kombinera dessa metoder med dataflödespipelines eller nätverk genom att skicka indata till huvudnoden (ett målblock) och genom att ta emot utdata från terminalnoden i pipelinen eller terminalnoderna i nätverket (ett eller flera källblock). Du kan också använda Choose metoden för att läsa från den första av de angivna källorna som har tillgängliga data och utföra åtgärder på dessa data.
Källblock erbjuder data till målblock genom att anropa ITargetBlock<TInput>.OfferMessage metoden. Målblocket svarar på ett meddelande som erbjuds på något av tre sätt: det kan acceptera meddelandet, avvisa meddelandet eller skjuta upp meddelandet. När målet accepterar meddelandet OfferMessage returnerar Acceptedmetoden . När målet nekar meddelandet OfferMessage returnerar Declinedmetoden . När målet kräver att det inte längre tar emot några meddelanden från källan OfferMessage returnerar DecliningPermanently. De fördefinierade typerna av källblock erbjuder inte meddelanden till länkade mål när ett sådant returvärde har tagits emot och de tar automatiskt bort länk från sådana mål.
När ett målblock skjuter upp meddelandet för senare användning OfferMessage returnerar Postponedmetoden . Ett målblock som skjuter upp ett meddelande kan senare anropa ISourceBlock<TOutput>.ReserveMessage metoden för att försöka reservera det erbjudna meddelandet. I det här läget är meddelandet antingen fortfarande tillgängligt och kan användas av målblocket, eller så har meddelandet tagits av ett annat mål. När målblocket senare kräver meddelandet eller inte längre behöver meddelandet anropas ISourceBlock<TOutput>.ConsumeMessage metoden eller ReleaseReservation . Meddelandereservation används vanligtvis av de typer av dataflödesblock som fungerar i icke-girigt läge. Icke-girigt läge förklaras senare i det här dokumentet. I stället för att reservera ett uppskjutet ISourceBlock<TOutput>.ConsumeMessage meddelande kan ett målblock också använda metoden för att försöka använda det uppskjutna meddelandet direkt.
Slutförande av dataflödesblock
Dataflödesblock stöder också begreppet slutförande. Ett dataflödesblock som är i slutfört tillstånd utför inte något ytterligare arbete. Varje dataflödesblock har ett associerat System.Threading.Tasks.Task objekt, som kallas för en slutförandeaktivitet, som representerar slutförandestatusen för blocket. Eftersom du kan vänta tills ett Task objekt har slutförts kan du med hjälp av slutförandeuppgifter vänta tills en eller flera terminalnoder i ett dataflödesnätverk har slutförts. Gränssnittet IDataflowBlock definierar Complete metoden, som informerar dataflödesblocket för en begäran om att den ska slutföras, och Completion egenskapen som returnerar slutförandeaktiviteten för dataflödesblocket. Både ISourceBlock<TOutput> och ITargetBlock<TInput> ärver IDataflowBlock gränssnittet.
Det finns två sätt att avgöra om ett dataflödesblock slutfördes utan fel, påträffade ett eller flera fel eller avbröts. Det första sättet är att anropa Task.Wait metoden för slutförandeaktiviteten i ett trycatch-block (Try-Catch i Visual Basic). I följande exempel skapas ett ActionBlock<TInput> objekt som genererar ArgumentOutOfRangeException om dess indatavärde är mindre än noll. AggregateException utlöses när det här exemplet anropar Wait slutförandeaktiviteten. ArgumentOutOfRangeException Nås via InnerExceptions -egenskapen för AggregateException objektet.
// Create an ActionBlock<int> object that prints its input
// and throws ArgumentOutOfRangeException if the input
// is less than zero.
var throwIfNegative = new ActionBlock<int>(n =>
{
Console.WriteLine("n = {0}", n);
if (n < 0)
{
throw new ArgumentOutOfRangeException();
}
});
// Post values to the block.
throwIfNegative.Post(0);
throwIfNegative.Post(-1);
throwIfNegative.Post(1);
throwIfNegative.Post(-2);
throwIfNegative.Complete();
// Wait for completion in a try/catch block.
try
{
throwIfNegative.Completion.Wait();
}
catch (AggregateException ae)
{
// If an unhandled exception occurs during dataflow processing, all
// exceptions are propagated through an AggregateException object.
ae.Handle(e =>
{
Console.WriteLine("Encountered {0}: {1}",
e.GetType().Name, e.Message);
return true;
});
}
/* Output:
n = 0
n = -1
Encountered ArgumentOutOfRangeException: Specified argument was out of the range
of valid values.
*/
' Create an ActionBlock<int> object that prints its input
' and throws ArgumentOutOfRangeException if the input
' is less than zero.
Dim throwIfNegative = New ActionBlock(Of Integer)(Sub(n)
Console.WriteLine("n = {0}", n)
If n < 0 Then
Throw New ArgumentOutOfRangeException()
End If
End Sub)
' Post values to the block.
throwIfNegative.Post(0)
throwIfNegative.Post(-1)
throwIfNegative.Post(1)
throwIfNegative.Post(-2)
throwIfNegative.Complete()
' Wait for completion in a try/catch block.
Try
throwIfNegative.Completion.Wait()
Catch ae As AggregateException
' If an unhandled exception occurs during dataflow processing, all
' exceptions are propagated through an AggregateException object.
ae.Handle(Function(e)
Console.WriteLine("Encountered {0}: {1}", e.GetType().Name, e.Message)
Return True
End Function)
End Try
' Output:
' n = 0
' n = -1
' Encountered ArgumentOutOfRangeException: Specified argument was out of the range
' of valid values.
'
Det här exemplet visar i vilket fall ett undantag inte hanteras i ombudet för ett körningsdataflödesblock. Vi rekommenderar att du hanterar undantag i sådana block. Men om du inte kan göra det fungerar blocket som om det avbröts och bearbetar inte inkommande meddelanden.
När ett dataflödesblock uttryckligen AggregateException avbryts innehåller OperationCanceledException objektet i egenskapen InnerExceptions . Mer information om annullering av dataflöden finns i avsnittet Aktivera annullering .
Det andra sättet att fastställa slutförandestatusen för ett dataflödesblock är att använda en fortsättning av slutförandeaktiviteten, eller att använda asynkrona språkfunktioner i C# och Visual Basic för att asynkront vänta på slutförandeaktiviteten. Ombudet som du anger för Task.ContinueWith metoden tar ett Task objekt som representerar den föregående uppgiften. När det gäller Completion egenskapen tar ombudet för fortsättningen själva slutförandeuppgiften. Följande exempel liknar det föregående, förutom att det även använder ContinueWith metoden för att skapa en fortsättningsaktivitet som skriver ut statusen för den övergripande dataflödesåtgärden.
// Create an ActionBlock<int> object that prints its input
// and throws ArgumentOutOfRangeException if the input
// is less than zero.
var throwIfNegative = new ActionBlock<int>(n =>
{
Console.WriteLine("n = {0}", n);
if (n < 0)
{
throw new ArgumentOutOfRangeException();
}
});
// Create a continuation task that prints the overall
// task status to the console when the block finishes.
throwIfNegative.Completion.ContinueWith(task =>
{
Console.WriteLine("The status of the completion task is '{0}'.",
task.Status);
});
// Post values to the block.
throwIfNegative.Post(0);
throwIfNegative.Post(-1);
throwIfNegative.Post(1);
throwIfNegative.Post(-2);
throwIfNegative.Complete();
// Wait for completion in a try/catch block.
try
{
throwIfNegative.Completion.Wait();
}
catch (AggregateException ae)
{
// If an unhandled exception occurs during dataflow processing, all
// exceptions are propagated through an AggregateException object.
ae.Handle(e =>
{
Console.WriteLine("Encountered {0}: {1}",
e.GetType().Name, e.Message);
return true;
});
}
/* Output:
n = 0
n = -1
The status of the completion task is 'Faulted'.
Encountered ArgumentOutOfRangeException: Specified argument was out of the range
of valid values.
*/
' Create an ActionBlock<int> object that prints its input
' and throws ArgumentOutOfRangeException if the input
' is less than zero.
Dim throwIfNegative = New ActionBlock(Of Integer)(Sub(n)
Console.WriteLine("n = {0}", n)
If n < 0 Then
Throw New ArgumentOutOfRangeException()
End If
End Sub)
' Create a continuation task that prints the overall
' task status to the console when the block finishes.
throwIfNegative.Completion.ContinueWith(Sub(task) Console.WriteLine("The status of the completion task is '{0}'.", task.Status))
' Post values to the block.
throwIfNegative.Post(0)
throwIfNegative.Post(-1)
throwIfNegative.Post(1)
throwIfNegative.Post(-2)
throwIfNegative.Complete()
' Wait for completion in a try/catch block.
Try
throwIfNegative.Completion.Wait()
Catch ae As AggregateException
' If an unhandled exception occurs during dataflow processing, all
' exceptions are propagated through an AggregateException object.
ae.Handle(Function(e)
Console.WriteLine("Encountered {0}: {1}", e.GetType().Name, e.Message)
Return True
End Function)
End Try
' Output:
' n = 0
' n = -1
' The status of the completion task is 'Faulted'.
' Encountered ArgumentOutOfRangeException: Specified argument was out of the range
' of valid values.
'
Du kan också använda egenskaper som IsCanceled i brödtexten i fortsättningsaktiviteten för att fastställa ytterligare information om slutförandestatusen för ett dataflödesblock. Mer information om fortsättningsuppgifter och hur de relaterar till annullering och felhantering finns i Länka uppgifter med hjälp av fortsättningsuppgifter, aktivitetsavbokning och undantagshantering.
Fördefinierade typer av dataflödesblock
TPL-dataflödesbiblioteket innehåller flera fördefinierade blocktyper för dataflöden. Dessa typer är indelade i tre kategorier: buffringsblock, körningsblock och grupperingsblock. I följande avsnitt beskrivs de blocktyper som utgör dessa kategorier.
Buffringsblock
Buffringsblock innehåller data för användning av datakonsumenter. TPL-dataflödesbiblioteket innehåller tre typer av buffertblock: System.Threading.Tasks.Dataflow.BufferBlock<T>, System.Threading.Tasks.Dataflow.BroadcastBlock<T>och System.Threading.Tasks.Dataflow.WriteOnceBlock<T>.
BufferBlock<T>
Klassen BufferBlock<T> representerar en asynkron meddelandestruktur för generell användning. Den här klassen lagrar en fifo-kö (first in, first out) som kan skrivas till av flera källor eller läsas från av flera mål. När ett mål tar emot ett meddelande från ett BufferBlock<T> objekt tas meddelandet bort från meddelandekön. Även om ett BufferBlock<T> objekt kan ha flera mål får därför bara ett mål varje meddelande. Klassen BufferBlock<T> är användbar när du vill skicka flera meddelanden till en annan komponent och komponenten måste ta emot varje meddelande.
I följande grundläggande exempel skickas flera Int32 värden till ett BufferBlock<T> objekt och dessa värden läss sedan tillbaka från objektet.
// Create a BufferBlock<int> object.
var bufferBlock = new BufferBlock<int>();
// Post several messages to the block.
for (int i = 0; i < 3; i++)
{
bufferBlock.Post(i);
}
// Receive the messages back from the block.
for (int i = 0; i < 3; i++)
{
Console.WriteLine(bufferBlock.Receive());
}
/* Output:
0
1
2
*/
' Create a BufferBlock<int> object.
Dim bufferBlock = New BufferBlock(Of Integer)()
' Post several messages to the block.
For i As Integer = 0 To 2
bufferBlock.Post(i)
Next i
' Receive the messages back from the block.
For i As Integer = 0 To 2
Console.WriteLine(bufferBlock.Receive())
Next i
' Output:
' 0
' 1
' 2
'
Ett fullständigt exempel som visar hur du skriver meddelanden till och läser meddelanden från ett BufferBlock<T> objekt finns i Så här skriver du meddelanden till och läser meddelanden från ett dataflödesblock.
BroadcastBlock<T>
Klassen BroadcastBlock<T> är användbar när du måste skicka flera meddelanden till en annan komponent, men komponenten behöver bara det senaste värdet. Den här klassen är också användbar när du vill sända ett meddelande till flera komponenter.
I följande grundläggande exempel skickas ett Double värde till ett BroadcastBlock<T> objekt och värdet läss sedan tillbaka från objektet flera gånger. Eftersom värden inte tas bort från BroadcastBlock<T> objekt när de har lästs är samma värde tillgängligt varje gång.
// Create a BroadcastBlock<double> object.
var broadcastBlock = new BroadcastBlock<double>(null);
// Post a message to the block.
broadcastBlock.Post(Math.PI);
// Receive the messages back from the block several times.
for (int i = 0; i < 3; i++)
{
Console.WriteLine(broadcastBlock.Receive());
}
/* Output:
3.14159265358979
3.14159265358979
3.14159265358979
*/
' Create a BroadcastBlock<double> object.
Dim broadcastBlock = New BroadcastBlock(Of Double)(Nothing)
' Post a message to the block.
broadcastBlock.Post(Math.PI)
' Receive the messages back from the block several times.
For i As Integer = 0 To 2
Console.WriteLine(broadcastBlock.Receive())
Next i
' Output:
' 3.14159265358979
' 3.14159265358979
' 3.14159265358979
'
Ett fullständigt exempel som visar hur du använder BroadcastBlock<T> för att sända ett meddelande till flera målblock finns i Så här anger du en schemaläggare i ett dataflödesblock.
WriteOnceBlock<T>
Klassen WriteOnceBlock<T> liknar BroadcastBlock<T> klassen, förutom att ett WriteOnceBlock<T> objekt endast kan skrivas till en gång. Du kan tänka WriteOnceBlock<T> dig att det liknar nyckelordet C# readonly (ReadOnly i Visual Basic), förutom att ett WriteOnceBlock<T> objekt blir oföränderligt när det har fått ett värde i stället för vid konstruktion. BroadcastBlock<T> Precis som klassen tas meddelandet inte bort från objektet när ett mål tar emot ett meddelande från ett WriteOnceBlock<T> objekt. Därför får flera mål en kopia av meddelandet. Klassen WriteOnceBlock<T> är användbar när du bara vill sprida den första av flera meddelanden.
Följande grundläggande exempel publicerar flera String värden i ett WriteOnceBlock<T> objekt och läser sedan tillbaka värdet från objektet. Eftersom ett WriteOnceBlock<T> objekt endast kan skrivas till en gång, när ett WriteOnceBlock<T> objekt har fått ett meddelande, ignoreras efterföljande meddelanden.
// Create a WriteOnceBlock<string> object.
var writeOnceBlock = new WriteOnceBlock<string>(null);
// Post several messages to the block in parallel. The first
// message to be received is written to the block.
// Subsequent messages are discarded.
Parallel.Invoke(
() => writeOnceBlock.Post("Message 1"),
() => writeOnceBlock.Post("Message 2"),
() => writeOnceBlock.Post("Message 3"));
// Receive the message from the block.
Console.WriteLine(writeOnceBlock.Receive());
/* Sample output:
Message 2
*/
' Create a WriteOnceBlock<string> object.
Dim writeOnceBlock = New WriteOnceBlock(Of String)(Nothing)
' Post several messages to the block in parallel. The first
' message to be received is written to the block.
' Subsequent messages are discarded.
Parallel.Invoke(Function() writeOnceBlock.Post("Message 1"), Function() writeOnceBlock.Post("Message 2"), Function() writeOnceBlock.Post("Message 3"))
' Receive the message from the block.
Console.WriteLine(writeOnceBlock.Receive())
' Sample output:
' Message 2
'
Ett fullständigt exempel som visar hur du använder WriteOnceBlock<T> för att ta emot värdet för den första åtgärden som slutförs finns i How to: Unlink Dataflow Blocks (Så här tar du bort länk till dataflödesblock).
Körningsblock
Körningsblock anropar en delegerad användare för varje del av mottagna data. TPL-dataflödesbiblioteket innehåller tre typer av körningsblock: ActionBlock<TInput>, System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput>och System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput>.
ActionBlock<T>
Klassen ActionBlock<TInput> är ett målblock som anropar ett ombud när den tar emot data. Tänk på ett ActionBlock<TInput> objekt som ett ombud som körs asynkront när data blir tillgängliga. Ombudet som du anger för ett ActionBlock<TInput> objekt kan vara av typen Action<T> eller typen System.Func<TInput, Task>. När du använder ett ActionBlock<TInput> objekt med Action<T>anses bearbetningen av varje indataelement slutföras när ombudet returnerar. När du använder ett ActionBlock<TInput> objekt med System.Func<TInput, Task>anses bearbetningen av varje indataelement endast slutföras när det returnerade Task objektet har slutförts. Genom att använda dessa två mekanismer kan du använda ActionBlock<TInput> för både synkron och asynkron bearbetning av varje indataelement.
I följande grundläggande exempel postas flera Int32 värden till ett ActionBlock<TInput> objekt. Objektet ActionBlock<TInput> skriver ut dessa värden till konsolen. Det här exemplet anger sedan blocket till slutfört tillstånd och väntar tills alla dataflödesuppgifter har slutförts.
// Create an ActionBlock<int> object that prints values
// to the console.
var actionBlock = new ActionBlock<int>(n => Console.WriteLine(n));
// Post several messages to the block.
for (int i = 0; i < 3; i++)
{
actionBlock.Post(i * 10);
}
// Set the block to the completed state and wait for all
// tasks to finish.
actionBlock.Complete();
actionBlock.Completion.Wait();
/* Output:
0
10
20
*/
' Create an ActionBlock<int> object that prints values
' to the console.
Dim actionBlock = New ActionBlock(Of Integer)(Function(n) WriteLine(n))
' Post several messages to the block.
For i As Integer = 0 To 2
actionBlock.Post(i * 10)
Next i
' Set the block to the completed state and wait for all
' tasks to finish.
actionBlock.Complete()
actionBlock.Completion.Wait()
' Output:
' 0
' 10
' 20
'
Fullständiga exempel som visar hur du använder ombud med klassen finns i ActionBlock<TInput>Så här utför du åtgärder när ett dataflödesblock tar emot data.
TransformBlock<TInput, TOutput>
Klassen TransformBlock<TInput,TOutput> liknar klassen, förutom att den ActionBlock<TInput> fungerar både som källa och som mål. Ombudet som du skickar till ett TransformBlock<TInput,TOutput> objekt returnerar ett värde av typen TOutput. Ombudet som du anger för ett TransformBlock<TInput,TOutput> objekt kan vara av typen System.Func<TInput, TOutput> eller typen System.Func<TInput, Task<TOutput>>. När du använder ett TransformBlock<TInput,TOutput> objekt med System.Func<TInput, TOutput>anses bearbetningen av varje indataelement slutföras när ombudet returnerar. När du använder ett TransformBlock<TInput,TOutput> objekt som används med System.Func<TInput, Task<TOutput>>anses bearbetningen av varje indataelement endast slutföras när det returnerade objektet har slutförts Task<TResult> . Precis som med ActionBlock<TInput>kan TransformBlock<TInput,TOutput> du använda dessa två mekanismer för både synkron och asynkron bearbetning av varje indataelement.
I följande grundläggande exempel skapas ett TransformBlock<TInput,TOutput> objekt som beräknar kvadratroten för dess indata. Objektet TransformBlock<TInput,TOutput> tar Int32 värden som indata och genererar Double värden som utdata.
// Create a TransformBlock<int, double> object that
// computes the square root of its input.
var transformBlock = new TransformBlock<int, double>(n => Math.Sqrt(n));
// Post several messages to the block.
transformBlock.Post(10);
transformBlock.Post(20);
transformBlock.Post(30);
// Read the output messages from the block.
for (int i = 0; i < 3; i++)
{
Console.WriteLine(transformBlock.Receive());
}
/* Output:
3.16227766016838
4.47213595499958
5.47722557505166
*/
' Create a TransformBlock<int, double> object that
' computes the square root of its input.
Dim transformBlock = New TransformBlock(Of Integer, Double)(Function(n) Math.Sqrt(n))
' Post several messages to the block.
transformBlock.Post(10)
transformBlock.Post(20)
transformBlock.Post(30)
' Read the output messages from the block.
For i As Integer = 0 To 2
Console.WriteLine(transformBlock.Receive())
Next i
' Output:
' 3.16227766016838
' 4.47213595499958
' 5.47722557505166
'
Fullständiga exempel som används TransformBlock<TInput,TOutput> i ett nätverk av dataflödesblock som utför bildbearbetning i ett Windows Forms-program finns i Genomgång: Använda dataflöde i ett Windows Forms-program.
TransformManyBlock<TInput, TOutput>
Klassen TransformManyBlock<TInput,TOutput> liknar TransformBlock<TInput,TOutput> klassen, förutom att TransformManyBlock<TInput,TOutput> genererar noll eller fler utdatavärden för varje indatavärde, i stället för endast ett utdatavärde för varje indatavärde. Ombudet som du anger för ett TransformManyBlock<TInput,TOutput> objekt kan vara av typen System.Func<TInput, IEnumerable<TOutput>> eller typen System.Func<TInput, Task<IEnumerable<TOutput>>>. När du använder ett TransformManyBlock<TInput,TOutput> objekt med System.Func<TInput, IEnumerable<TOutput>>anses bearbetningen av varje indataelement slutföras när ombudet returnerar. När du använder ett TransformManyBlock<TInput,TOutput> objekt med System.Func<TInput, Task<IEnumerable<TOutput>>>betraktas bearbetningen av varje indataelement som fullständig endast när det returnerade System.Threading.Tasks.Task<IEnumerable<TOutput>> objektet har slutförts.
I följande grundläggande exempel skapas ett TransformManyBlock<TInput,TOutput> objekt som delar upp strängar i sina enskilda teckensekvenser. Objektet TransformManyBlock<TInput,TOutput> tar String värden som indata och genererar Char värden som utdata.
// Create a TransformManyBlock<string, char> object that splits
// a string into its individual characters.
var transformManyBlock = new TransformManyBlock<string, char>(
s => s.ToCharArray());
// Post two messages to the first block.
transformManyBlock.Post("Hello");
transformManyBlock.Post("World");
// Receive all output values from the block.
for (int i = 0; i < ("Hello" + "World").Length; i++)
{
Console.WriteLine(transformManyBlock.Receive());
}
/* Output:
H
e
l
l
o
W
o
r
l
d
*/
' Create a TransformManyBlock<string, char> object that splits
' a string into its individual characters.
Dim transformManyBlock = New TransformManyBlock(Of String, Char)(Function(s) s.ToCharArray())
' Post two messages to the first block.
transformManyBlock.Post("Hello")
transformManyBlock.Post("World")
' Receive all output values from the block.
For i As Integer = 0 To ("Hello" & "World").Length - 1
Console.WriteLine(transformManyBlock.Receive())
Next i
' Output:
' H
' e
' l
' l
' o
' W
' o
' r
' l
' d
'
Fullständiga exempel som används TransformManyBlock<TInput,TOutput> för att skapa flera oberoende utdata för varje indata i en dataflödespipeline finns i Genomgång: Skapa en dataflödespipeline.
Grad av parallellitet
Varje ActionBlock<TInput>, TransformBlock<TInput,TOutput>och TransformManyBlock<TInput,TOutput> -objekt buffrar indatameddelanden tills blocket är redo att bearbeta dem. Som standard bearbetar dessa klasser meddelanden i den ordning de tas emot, ett meddelande i taget. Du kan också ange graden av parallellitet för att aktivera ActionBlock<TInput>och TransformBlock<TInput,TOutput>TransformManyBlock<TInput,TOutput> objekt för att bearbeta flera meddelanden samtidigt. Mer information om samtidig körning finns i avsnittet Ange graden av parallellitet senare i det här dokumentet. Ett exempel som anger graden av parallellitet för att göra det möjligt för ett körningsdataflödesblock att bearbeta fler än ett meddelande i taget finns i Så här anger du graden av parallellitet i ett dataflödesblock.
Sammanfattning av ombudstyper
I följande tabell sammanfattas de ombudstyper som du kan ange till ActionBlock<TInput>, TransformBlock<TInput,TOutput>och TransformManyBlock<TInput,TOutput> objekt. Den här tabellen anger också om ombudstypen fungerar synkront eller asynkront.
| Typ | Synkron ombudstyp | Asynkron ombudstyp |
|---|---|---|
| ActionBlock<TInput> | System.Action |
System.Func<TInput, Task> |
| TransformBlock<TInput,TOutput> | System.Func<TInput, TOutput> |
System.Func<TInput, Task<TOutput>> |
| TransformManyBlock<TInput,TOutput> | System.Func<TInput, IEnumerable<TOutput>> |
System.Func<TInput, Task<IEnumerable<TOutput>>> |
Du kan också använda lambda-uttryck när du arbetar med körningsblocktyper. Ett exempel som visar hur du använder ett lambda-uttryck med ett körningsblock finns i Så här utför du åtgärder när ett dataflödesblock tar emot data.
Grupperingsblock
Grupperingsblock kombinerar data från en eller flera källor och under olika begränsningar. TPL-dataflödesbiblioteket innehåller tre typer av kopplingsblock: BatchBlock<T>, JoinBlock<T1,T2>och BatchedJoinBlock<T1,T2>.
BatchBlock<T>
Klassen BatchBlock<T> kombinerar uppsättningar med indata, som kallas batchar, till matriser med utdata. Du anger storleken på varje batch när du skapar ett BatchBlock<T> objekt. BatchBlock<T> När objektet tar emot det angivna antalet indataelement sprids asynkront ut en matris som innehåller dessa element. Om ett BatchBlock<T> objekt är inställt på det slutförda tillståndet men inte innehåller tillräckligt med element för att bilda en batch, sprids en slutlig matris som innehåller de återstående indataelementen.
Klassen BatchBlock<T> fungerar antingen i girigt eller icke-girigt läge. I girigt läge, som är standard, accepterar ett BatchBlock<T> objekt varje meddelande som det erbjuds och sprider ut en matris när det har fått det angivna antalet element. I icke-girigt läge skjuter ett BatchBlock<T> objekt upp alla inkommande meddelanden tills tillräckligt många källor har erbjudit meddelanden till blocket för att bilda en batch. Girigt läge presterar vanligtvis bättre än icke-girigt läge eftersom det kräver mindre bearbetningskostnader. Du kan dock använda icke-girigt läge när du måste samordna förbrukningen från flera källor på ett atomiskt sätt. Ange icke-girigt läge genom att False ange Greedy i parametern dataflowBlockOptions i BatchBlock<T> konstruktorn.
I följande grundläggande exempel postas flera Int32 värden till ett BatchBlock<T> objekt som innehåller tio element i en batch. För att garantera att alla värden sprids ut ur BatchBlock<T>anropas metoden i Complete det här exemplet. Metoden Complete anger BatchBlock<T> objektet till slutfört tillstånd och därför BatchBlock<T> sprider objektet ut eventuella återstående element som en slutlig batch.
// Create a BatchBlock<int> object that holds ten
// elements per batch.
var batchBlock = new BatchBlock<int>(10);
// Post several values to the block.
for (int i = 0; i < 13; i++)
{
batchBlock.Post(i);
}
// Set the block to the completed state. This causes
// the block to propagate out any remaining
// values as a final batch.
batchBlock.Complete();
// Print the sum of both batches.
Console.WriteLine("The sum of the elements in batch 1 is {0}.",
batchBlock.Receive().Sum());
Console.WriteLine("The sum of the elements in batch 2 is {0}.",
batchBlock.Receive().Sum());
/* Output:
The sum of the elements in batch 1 is 45.
The sum of the elements in batch 2 is 33.
*/
' Create a BatchBlock<int> object that holds ten
' elements per batch.
Dim batchBlock = New BatchBlock(Of Integer)(10)
' Post several values to the block.
For i As Integer = 0 To 12
batchBlock.Post(i)
Next i
' Set the block to the completed state. This causes
' the block to propagate out any remaining
' values as a final batch.
batchBlock.Complete()
' Print the sum of both batches.
Console.WriteLine("The sum of the elements in batch 1 is {0}.", batchBlock.Receive().Sum())
Console.WriteLine("The sum of the elements in batch 2 is {0}.", batchBlock.Receive().Sum())
' Output:
' The sum of the elements in batch 1 is 45.
' The sum of the elements in batch 2 is 33.
'
Ett fullständigt exempel som används BatchBlock<T> för att förbättra effektiviteten för databasinfogningsåtgärder finns i Genomgång: Använda BatchBlock och BatchedJoinBlock för att förbättra effektiviteten.
JoinBlock<T1, T2, ...>
Klasserna JoinBlock<T1,T2> och JoinBlock<T1,T2,T3> samlar indataelement och sprider ut System.Tuple<T1,T2> eller System.Tuple<T1,T2,T3> objekt som innehåller dessa element. Klasserna JoinBlock<T1,T2> och JoinBlock<T1,T2,T3> ärver inte från ITargetBlock<TInput>. I stället tillhandahåller de egenskaper, Target1, Target2och Target3, som implementerar ITargetBlock<TInput>.
Som BatchBlock<T>och JoinBlock<T1,T2>JoinBlock<T1,T2,T3> fungerar i antingen girigt eller icke-girigt läge. I girigt läge, som är standard, accepterar ett JoinBlock<T1,T2> eller JoinBlock<T1,T2,T3> -objekt varje meddelande som det erbjuds och sprider ut en tupplar efter att vart och ett av målen har fått minst ett meddelande. I icke-girigt läge skjuter ett JoinBlock<T1,T2> eller JoinBlock<T1,T2,T3> -objekt upp alla inkommande meddelanden tills alla mål har erbjudits de data som krävs för att skapa en tupplar. I det här läget deltar blocket i ett tvåfas-incheckningsprotokoll för att atomiskt hämta alla nödvändiga objekt från källorna. Denna senareläggning gör det möjligt för en annan entitet att använda data under tiden, så att det övergripande systemet kan göra framsteg framåt.
Följande grundläggande exempel visar ett fall där ett JoinBlock<T1,T2,T3> objekt kräver flera data för att beräkna ett värde. Det här exemplet skapar ett JoinBlock<T1,T2,T3> objekt som kräver två Int32 värden och ett Char värde för att utföra en aritmetikåtgärd.
// Create a JoinBlock<int, int, char> object that requires
// two numbers and an operator.
var joinBlock = new JoinBlock<int, int, char>();
// Post two values to each target of the join.
joinBlock.Target1.Post(3);
joinBlock.Target1.Post(6);
joinBlock.Target2.Post(5);
joinBlock.Target2.Post(4);
joinBlock.Target3.Post('+');
joinBlock.Target3.Post('-');
// Receive each group of values and apply the operator part
// to the number parts.
for (int i = 0; i < 2; i++)
{
var data = joinBlock.Receive();
switch (data.Item3)
{
case '+':
Console.WriteLine("{0} + {1} = {2}",
data.Item1, data.Item2, data.Item1 + data.Item2);
break;
case '-':
Console.WriteLine("{0} - {1} = {2}",
data.Item1, data.Item2, data.Item1 - data.Item2);
break;
default:
Console.WriteLine("Unknown operator '{0}'.", data.Item3);
break;
}
}
/* Output:
3 + 5 = 8
6 - 4 = 2
*/
' Create a JoinBlock<int, int, char> object that requires
' two numbers and an operator.
Dim joinBlock = New JoinBlock(Of Integer, Integer, Char)()
' Post two values to each target of the join.
joinBlock.Target1.Post(3)
joinBlock.Target1.Post(6)
joinBlock.Target2.Post(5)
joinBlock.Target2.Post(4)
joinBlock.Target3.Post("+"c)
joinBlock.Target3.Post("-"c)
' Receive each group of values and apply the operator part
' to the number parts.
For i As Integer = 0 To 1
Dim data = joinBlock.Receive()
Select Case data.Item3
Case "+"c
Console.WriteLine("{0} + {1} = {2}", data.Item1, data.Item2, data.Item1 + data.Item2)
Case "-"c
Console.WriteLine("{0} - {1} = {2}", data.Item1, data.Item2, data.Item1 - data.Item2)
Case Else
Console.WriteLine("Unknown operator '{0}'.", data.Item3)
End Select
Next i
' Output:
' 3 + 5 = 8
' 6 - 4 = 2
'
Ett fullständigt exempel som använder JoinBlock<T1,T2> objekt i icke-girigt läge för att gemensamt dela en resurs finns i Så här använder du JoinBlock för att läsa data från flera källor.
BatchedJoinBlock<T1, T2, ...>
Klasserna BatchedJoinBlock<T1,T2> och BatchedJoinBlock<T1,T2,T3> samlar in batchar med indataelement och sprider ut System.Tuple(IList(T1), IList(T2)) eller System.Tuple(IList(T1), IList(T2), IList(T3)) objekt som innehåller dessa element. Tänk på BatchedJoinBlock<T1,T2> som en kombination av BatchBlock<T> och JoinBlock<T1,T2>. Ange storleken på varje batch när du skapar ett BatchedJoinBlock<T1,T2> objekt. BatchedJoinBlock<T1,T2> innehåller också egenskaper och Target1Target2, som implementerar ITargetBlock<TInput>. När det angivna antalet indataelement tas emot från alla mål BatchedJoinBlock<T1,T2> sprider objektet asynkront ut ett System.Tuple(IList(T1), IList(T2)) objekt som innehåller dessa element.
I följande grundläggande exempel skapas ett BatchedJoinBlock<T1,T2> objekt som innehåller resultat, Int32 värden och fel som är Exception objekt. Det här exemplet utför flera åtgärder och skriver resultat till Target1 egenskapen och fel till Target2 egenskapen för BatchedJoinBlock<T1,T2> objektet. Eftersom antalet lyckade och misslyckade åtgärder är okänt i förväg gör objekten IList<T> att varje mål kan ta emot noll eller fler värden.
// For demonstration, create a Func<int, int> that
// returns its argument, or throws ArgumentOutOfRangeException
// if the argument is less than zero.
Func<int, int> DoWork = n =>
{
if (n < 0)
throw new ArgumentOutOfRangeException();
return n;
};
// Create a BatchedJoinBlock<int, Exception> object that holds
// seven elements per batch.
var batchedJoinBlock = new BatchedJoinBlock<int, Exception>(7);
// Post several items to the block.
foreach (int i in new int[] { 5, 6, -7, -22, 13, 55, 0 })
{
try
{
// Post the result of the worker to the
// first target of the block.
batchedJoinBlock.Target1.Post(DoWork(i));
}
catch (ArgumentOutOfRangeException e)
{
// If an error occurred, post the Exception to the
// second target of the block.
batchedJoinBlock.Target2.Post(e);
}
}
// Read the results from the block.
var results = batchedJoinBlock.Receive();
// Print the results to the console.
// Print the results.
foreach (int n in results.Item1)
{
Console.WriteLine(n);
}
// Print failures.
foreach (Exception e in results.Item2)
{
Console.WriteLine(e.Message);
}
/* Output:
5
6
13
55
0
Specified argument was out of the range of valid values.
Specified argument was out of the range of valid values.
*/
' For demonstration, create a Func<int, int> that
' returns its argument, or throws ArgumentOutOfRangeException
' if the argument is less than zero.
Dim DoWork As Func(Of Integer, Integer) = Function(n)
If n < 0 Then
Throw New ArgumentOutOfRangeException()
End If
Return n
End Function
' Create a BatchedJoinBlock<int, Exception> object that holds
' seven elements per batch.
Dim batchedJoinBlock = New BatchedJoinBlock(Of Integer, Exception)(7)
' Post several items to the block.
For Each i As Integer In New Integer() {5, 6, -7, -22, 13, 55, 0}
Try
' Post the result of the worker to the
' first target of the block.
batchedJoinBlock.Target1.Post(DoWork(i))
Catch e As ArgumentOutOfRangeException
' If an error occurred, post the Exception to the
' second target of the block.
batchedJoinBlock.Target2.Post(e)
End Try
Next i
' Read the results from the block.
Dim results = batchedJoinBlock.Receive()
' Print the results to the console.
' Print the results.
For Each n As Integer In results.Item1
Console.WriteLine(n)
Next n
' Print failures.
For Each e As Exception In results.Item2
Console.WriteLine(e.Message)
Next e
' Output:
' 5
' 6
' 13
' 55
' 0
' Specified argument was out of the range of valid values.
' Specified argument was out of the range of valid values.
'
Ett fullständigt exempel som använder BatchedJoinBlock<T1,T2> för att samla in både resultat och eventuella undantag som inträffar när programmet läser från en databas finns i Genomgång: Använda BatchBlock och BatchedJoinBlock för att förbättra effektiviteten.
Konfigurera beteende för dataflödesblock
Du kan aktivera ytterligare alternativ genom att tillhandahålla ett System.Threading.Tasks.Dataflow.DataflowBlockOptions objekt till konstruktorn av blocktyper för dataflöden. Dessa alternativ styr beteendet, till exempel schemaläggaren som hanterar den underliggande uppgiften och graden av parallellitet. Har DataflowBlockOptions också härledda typer som anger beteende som är specifikt för vissa typer av dataflödesblock. I följande tabell sammanfattas vilken alternativtyp som är associerad med varje typ av dataflödesblock.
Följande avsnitt innehåller ytterligare information om de viktiga typer av blockeringsalternativ för dataflöden som är tillgängliga via klasserna System.Threading.Tasks.Dataflow.DataflowBlockOptions, System.Threading.Tasks.Dataflow.ExecutionDataflowBlockOptionsoch System.Threading.Tasks.Dataflow.GroupingDataflowBlockOptions .
Ange schemaläggaren
Varje fördefinierat dataflödesblock använder schemaläggningsmekanismen för TPL-uppgifter för att utföra aktiviteter som att sprida data till ett mål, ta emot data från en källa och köra användardefinierade ombud när data blir tillgängliga. TaskScheduler är en abstrakt klass som representerar en schemaläggare som köar aktiviteter till trådar. Standarduppgiftsschemaläggaren, Default, använder ThreadPool klassen för att köa och köra arbete. Du kan åsidosätta standardaktivitetsschemaläggaren genom att ange TaskScheduler egenskapen när du skapar ett dataflödesblockobjekt.
När samma schemaläggare hanterar flera dataflödesblock kan det framtvinga principer i dem. Om till exempel flera dataflödesblock har konfigurerats för att rikta in sig på den exklusiva schemaläggaren för samma ConcurrentExclusiveSchedulerPair objekt, serialiseras allt arbete som körs över dessa block. På samma sätt, om dessa block är konfigurerade för att rikta in sig på den samtidiga schemaläggaren för samma ConcurrentExclusiveSchedulerPair objekt, och schemaläggaren är konfigurerad för att ha en maximal samtidighetsnivå, är allt arbete från dessa block begränsat till det antalet samtidiga åtgärder. Ett exempel som använder ConcurrentExclusiveSchedulerPair klassen för att göra det möjligt att utföra läsåtgärder parallellt, men skrivåtgärder endast utförs av alla andra åtgärder, finns i How to: Specify a Task Scheduler in a Dataflow Block (Så här anger du en schemaläggare i ett dataflödesblock). Mer information om schemaläggare i TPL finns i klassavsnittet TaskScheduler .
Ange graden av parallellitet
Som standard bearbetar de tre körningsblocktyperna som TPL-dataflödesbiblioteket tillhandahåller, ActionBlock<TInput>, TransformBlock<TInput,TOutput>och TransformManyBlock<TInput,TOutput>, ett meddelande i taget. Dessa typer av dataflödesblock bearbetar också meddelanden i den ordning de tas emot. Om du vill aktivera dessa dataflödesblock för att bearbeta meddelanden samtidigt anger ExecutionDataflowBlockOptions.MaxDegreeOfParallelism du egenskapen när du skapar dataflödesblockobjektet.
Standardvärdet MaxDegreeOfParallelism är 1, vilket garanterar att dataflödesblocket bearbetar ett meddelande i taget. Om du anger den här egenskapen till ett värde som är större än 1 kan dataflödesblocket bearbeta flera meddelanden samtidigt. Om du anger den här egenskapen till DataflowBlockOptions.Unbounded kan den underliggande schemaläggaren hantera maximal samtidighetsgrad.
Viktigt!
När du anger en maximal grad av parallellitet som är större än 1 bearbetas flera meddelanden samtidigt och därför bearbetas kanske inte meddelanden i den ordning de tas emot. Den ordning i vilken meddelandena matas ut från blocket är dock samma som de tas emot i.
Eftersom egenskapen MaxDegreeOfParallelism representerar den maximala graden av parallellitet kan dataflödesblocket köras med en mindre grad av parallellitet än du anger. Dataflödesblocket kan använda en mindre grad av parallellitet för att uppfylla funktionskraven eller för att det saknas tillgängliga systemresurser. Ett dataflödesblock väljer aldrig mer parallellitet än du anger.
Värdet för MaxDegreeOfParallelism egenskapen är exklusivt för varje dataflödesblockobjekt. Om till exempel fyra blockobjekt för dataflöden var och en anger 1 för den maximala graden av parallellitet, kan alla fyra dataflödesblockobjekt potentiellt köras parallellt.
Ett exempel som anger den maximala graden av parallellitet så att långa åtgärder kan utföras parallellt finns i Så här anger du graden av parallellitet i ett dataflödesblock.
Ange antalet meddelanden per aktivitet
De fördefinierade blocktyperna för dataflöden använder uppgifter för att bearbeta flera indataelement. Detta hjälper till att minimera antalet aktivitetsobjekt som krävs för att bearbeta data, vilket gör att program kan köras mer effektivt. Men när aktiviteterna från en uppsättning dataflödesblock bearbetar data kan uppgifter från andra dataflödesblock behöva vänta på bearbetningstiden genom att köa meddelanden. Ange egenskapen för att ge bättre rättvisa mellan dataflödesuppgifter MaxMessagesPerTask . När MaxMessagesPerTask är inställt på DataflowBlockOptions.Unbounded, vilket är standard, bearbetar uppgiften som används av ett dataflödesblock så många meddelanden som är tillgängliga. När MaxMessagesPerTask är inställt på ett annat värde än Unboundedbearbetar dataflödesblocket högst det här antalet meddelanden per Task objekt. Även om inställningen av MaxMessagesPerTask egenskapen kan öka rättvisan mellan aktiviteter kan det leda till att systemet skapar fler uppgifter än vad som är nödvändigt, vilket kan minska prestandan.
Aktivera annullering
TPL tillhandahåller en mekanism som gör det möjligt för uppgifter att samordna annulleringen på ett samarbetsmässigt sätt. Ange egenskapen för att aktivera dataflödesblock för att delta i den här annulleringsmekanismen CancellationToken . När det här CancellationToken objektet är inställt på det avbrutna tillståndet blockeras alla dataflöden som övervakar den här tokens slutförande av det aktuella objektet, men som inte börjar bearbeta efterföljande objekt. Dessa dataflödesblock rensar även alla buffrade meddelanden, frigör anslutningar till alla käll- och målblock och övergår till det avbrutna tillståndet. Genom att övergå till det avbrutna Completion tillståndet har Status egenskapen angetts till Canceled, såvida inte ett undantag inträffade under bearbetningen. I så fall Status anges till Faulted.
Ett exempel som visar hur du använder annullering i ett Windows Forms-program finns i Så här: Avbryt ett dataflödesblock. Mer information om annullering i TPL finns i Annullering av aktiviteter.
Ange girigt kontra icke-girigt beteende
Flera grupperingsdataflödesblocktyper kan fungera i antingen girigt eller icke-girigt läge. Som standard fungerar de fördefinierade typerna av dataflödesblock i girigt läge.
För kopplingsblocktyper som JoinBlock<T1,T2>, innebär girigt läge att blocket omedelbart accepterar data även om motsvarande data som ska kopplas till ännu inte är tillgängliga. Icke-girigt läge innebär att blocket skjuter upp alla inkommande meddelanden tills ett är tillgängligt på vart och ett av dess mål för att slutföra kopplingen. Om något av de uppskjutna meddelandena inte längre är tillgängliga släpper kopplingsblocket alla uppskjutna meddelanden och startar om processen. BatchBlock<T> För klassen är girigt och icke-girigt beteende liknande, förutom att under icke-girigt läge skjuter ett BatchBlock<T> objekt upp alla inkommande meddelanden tills tillräckligt många är tillgängliga från olika källor för att slutföra en batch.
Om du vill ange icke-girigt läge för ett dataflödesblock anger du Greedy till False. Ett exempel som visar hur du använder icke-girigt läge för att aktivera flera kopplingsblock för att dela en datakälla mer effektivt finns i Så här använder du JoinBlock för att läsa data från flera källor.
Anpassade dataflödesblock
Även om TPL-dataflödesbiblioteket innehåller många fördefinierade blocktyper kan du skapa ytterligare blocktyper som utför anpassat beteende. Implementera gränssnitten ISourceBlock<TOutput> eller ITargetBlock<TInput> direkt eller använd Encapsulate metoden för att skapa ett komplext block som kapslar in beteendet för befintliga blocktyper. Exempel som visar hur du implementerar anpassade funktioner för dataflödesblock finns i Genomgång: Skapa en anpassad dataflödesblocktyp.
Närliggande information
| Title | Description |
|---|---|
| Anvisningar: Skriva meddelanden till och läsa meddelanden från ett dataflödesblock | Visar hur du skriver meddelanden till och läser meddelanden från ett BufferBlock<T> objekt. |
| Så här implementerar du ett mönster för producent-konsument-dataflöde | Beskriver hur du använder dataflödesmodellen för att implementera ett mönster för producent-konsument, där producenten skickar meddelanden till ett dataflödesblock och konsumenten läser meddelanden från det blocket. |
| Gör så här: Utför åtgärd när ett dataflödesblock tar emot data | Beskriver hur du tillhandahåller ombud för blocktyperna för körningsdataflöden, ActionBlock<TInput>, TransformBlock<TInput,TOutput>och TransformManyBlock<TInput,TOutput>. |
| Genomgång: Skapa en dataflödespipeline | Beskriver hur du skapar en dataflödespipeline som laddar ned text från webben och utför åtgärder för texten. |
| Så här gör du: Ta bort länk till dataflödesblock | Visar hur du använder LinkTo metoden för att ta bort länkning av ett målblock från källan när källan har ett meddelande till målet. |
| Genomgång: Använda dataflöde i ett Windows Forms-program | Visar hur du skapar ett nätverk av dataflödesblock som utför bildbearbetning i ett Windows Forms-program. |
| Gör så här: Avbryt ett dataflödesblock | Visar hur du använder annullering i ett Windows Forms-program. |
| Anvisningar: Använda JoinBlock för att läsa data från flera källor | Förklarar hur du använder JoinBlock<T1,T2> klassen för att utföra en åtgärd när data är tillgängliga från flera källor och hur du använder icke-girigt läge för att aktivera flera kopplingsblock för att dela en datakälla mer effektivt. |
| Anvisningar: Ange graden av parallellitet i ett dataflödesblock | Beskriver hur du ställer in MaxDegreeOfParallelism egenskapen så att ett körningsdataflödesblock kan bearbeta mer än ett meddelande i taget. |
| Anvisningar: Ange en schemaläggare i ett dataflödesblock | Visar hur du associerar en specifik schemaläggare när du använder dataflöde i ditt program. |
| Genomgång: Använda BatchBlock och BatchedJoinBlock för att förbättra effektiviteten | Beskriver hur du använder BatchBlock<T> klassen för att förbättra effektiviteten för databasinfogningsåtgärder och hur du använder BatchedJoinBlock<T1,T2> klassen för att samla in både resultaten och eventuella undantag som inträffar när programmet läser från en databas. |
| Genomgång: Skapa en anpassad dataflödesblocktyp | Visar två sätt att skapa en blocktyp för dataflöde som implementerar anpassat beteende. |
| Aktivitetsparallellt bibliotek (TPL) | Introducerar TPL, ett bibliotek som förenklar parallell och samtidig programmering i .NET Framework-program. |
Feedback
Kommer snart: Under hela 2024 kommer vi att fasa ut GitHub-problem som feedbackmekanism för innehåll och ersätta det med ett nytt feedbacksystem. Mer information finns i: https://aka.ms/ContentUserFeedback.
Skicka och visa feedback för