Procedure: Gegevensstroomblokken ontkoppelen
In dit document wordt beschreven hoe u een doelgegevensstroomblok loskoppelt van de bron.
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 drie TransformBlock<TInput,TOutput> objecten gemaakt, die de TrySolution methode aanroepen om een waarde te berekenen. In dit voorbeeld is alleen het resultaat van de eerste aanroep vereist om te TrySolution voltooien.
using System;
using System.Threading;
using System.Threading.Tasks.Dataflow;
// Demonstrates how to unlink dataflow blocks.
class DataflowReceiveAny
{
// Receives the value from the first provided source that has
// a message.
public static T ReceiveFromAny<T>(params ISourceBlock<T>[] sources)
{
// Create a WriteOnceBlock<T> object and link it to each source block.
var writeOnceBlock = new WriteOnceBlock<T>(e => e);
foreach (var source in sources)
{
// Setting MaxMessages to one instructs
// the source block to unlink from the WriteOnceBlock<T> object
// after offering the WriteOnceBlock<T> object one message.
source.LinkTo(writeOnceBlock, new DataflowLinkOptions { MaxMessages = 1 });
}
// Return the first value that is offered to the WriteOnceBlock object.
return writeOnceBlock.Receive();
}
// Demonstrates a function that takes several seconds to produce a result.
static int TrySolution(int n, CancellationToken ct)
{
// Simulate a lengthy operation that completes within three seconds
// or when the provided CancellationToken object is cancelled.
SpinWait.SpinUntil(() => ct.IsCancellationRequested,
new Random().Next(3000));
// Return a value.
return n + 42;
}
static void Main(string[] args)
{
// Create a shared CancellationTokenSource object to enable the
// TrySolution method to be cancelled.
var cts = new CancellationTokenSource();
// Create three TransformBlock<int, int> objects.
// Each TransformBlock<int, int> object calls the TrySolution method.
Func<int, int> action = n => TrySolution(n, cts.Token);
var trySolution1 = new TransformBlock<int, int>(action);
var trySolution2 = new TransformBlock<int, int>(action);
var trySolution3 = new TransformBlock<int, int>(action);
// Post data to each TransformBlock<int, int> object.
trySolution1.Post(11);
trySolution2.Post(21);
trySolution3.Post(31);
// Call the ReceiveFromAny<T> method to receive the result from the
// first TransformBlock<int, int> object to finish.
int result = ReceiveFromAny(trySolution1, trySolution2, trySolution3);
// Cancel all calls to TrySolution that are still active.
cts.Cancel();
// Print the result to the console.
Console.WriteLine("The solution is {0}.", result);
cts.Dispose();
}
}
/* Sample output:
The solution is 53.
*/
Imports System.Threading
Imports System.Threading.Tasks.Dataflow
' Demonstrates how to unlink dataflow blocks.
Friend Class DataflowReceiveAny
' Receives the value from the first provided source that has
' a message.
Public Shared Function ReceiveFromAny(Of T)(ParamArray ByVal sources() As ISourceBlock(Of T)) As T
' Create a WriteOnceBlock<T> object and link it to each source block.
Dim writeOnceBlock = New WriteOnceBlock(Of T)(Function(e) e)
For Each source In sources
' Setting MaxMessages to one instructs
' the source block to unlink from the WriteOnceBlock<T> object
' after offering the WriteOnceBlock<T> object one message.
source.LinkTo(writeOnceBlock, New DataflowLinkOptions With {.MaxMessages = 1})
Next source
' Return the first value that is offered to the WriteOnceBlock object.
Return writeOnceBlock.Receive()
End Function
' Demonstrates a function that takes several seconds to produce a result.
Private Shared Function TrySolution(ByVal n As Integer, ByVal ct As CancellationToken) As Integer
' Simulate a lengthy operation that completes within three seconds
' or when the provided CancellationToken object is cancelled.
SpinWait.SpinUntil(Function() ct.IsCancellationRequested, New Random().Next(3000))
' Return a value.
Return n + 42
End Function
Shared Sub Main(ByVal args() As String)
' Create a shared CancellationTokenSource object to enable the
' TrySolution method to be cancelled.
Dim cts = New CancellationTokenSource()
' Create three TransformBlock<int, int> objects.
' Each TransformBlock<int, int> object calls the TrySolution method.
Dim action As Func(Of Integer, Integer) = Function(n) TrySolution(n, cts.Token)
Dim trySolution1 = New TransformBlock(Of Integer, Integer)(action)
Dim trySolution2 = New TransformBlock(Of Integer, Integer)(action)
Dim trySolution3 = New TransformBlock(Of Integer, Integer)(action)
' Post data to each TransformBlock<int, int> object.
trySolution1.Post(11)
trySolution2.Post(21)
trySolution3.Post(31)
' Call the ReceiveFromAny<T> method to receive the result from the
' first TransformBlock<int, int> object to finish.
Dim result As Integer = ReceiveFromAny(trySolution1, trySolution2, trySolution3)
' Cancel all calls to TrySolution that are still active.
cts.Cancel()
' Print the result to the console.
Console.WriteLine("The solution is {0}.", result)
cts.Dispose()
End Sub
End Class
' Sample output:
'The solution is 53.
'
Als u de waarde wilt ontvangen van het eerste TransformBlock<TInput,TOutput> object dat is voltooid, definieert dit voorbeeld de ReceiveFromAny(T) methode. De ReceiveFromAny(T) methode accepteert een matrix met ISourceBlock<TOutput> objecten en koppelt elk van deze objecten aan een WriteOnceBlock<T> object. Wanneer u de LinkTo methode gebruikt om een brongegevensstroomblok te koppelen aan een doelblok, worden berichten door de bron naar het doel doorgegeven wanneer gegevens beschikbaar komen. Omdat de WriteOnceBlock<T> klasse alleen het eerste bericht accepteert dat het wordt aangeboden, produceert de ReceiveFromAny(T) methode het resultaat door de methode aan te Receive roepen. Hiermee wordt het eerste bericht gegenereerd dat aan het WriteOnceBlock<T> object wordt aangeboden. De LinkTo methode heeft een overbelaste versie die een DataflowLinkOptions object met een MaxMessages eigenschap gebruikt die, wanneer deze is ingesteld 1, het bronblok opdracht geeft om de koppeling van het doel op te heffen nadat het doel één bericht van de bron ontvangt. Het is belangrijk dat het WriteOnceBlock<T> object ontkoppelt van de bronnen omdat de relatie tussen de matrix van bronnen en het WriteOnceBlock<T> object niet meer nodig is nadat het WriteOnceBlock<T> object een bericht heeft ontvangen.
Als u wilt dat de resterende aanroepen worden TrySolution beëindigd nadat een van deze een waarde heeft berekend, neemt de TrySolution methode een CancellationToken object dat wordt geannuleerd nadat de aanroep om terug te ReceiveFromAny(T) keren is geannuleerd. De SpinUntil methode retourneert wanneer dit CancellationToken object wordt geannuleerd.
