Vorgehensweise: Aufheben der Verknüpfung von Datenflussblöcken

In diesem Dokument wird beschrieben, wie Sie die Verknüpfung eines Zieldatenflussblocks von der Quelle aufheben.

Hinweis

Die TPL-Datenflussbibliothek (der System.Threading.Tasks.Dataflow Namespace) ist in .NET 6 und höheren Versionen enthalten. Für .NET Framework- und .NET Standard-Projekte müssen Sie das 📦 NuGet-Paket "System.Threading.Tasks.Dataflow" installieren.

Example

Im folgenden Beispiel werden drei TransformBlock<TInput,TOutput> Objekte erstellt, von denen jede die TrySolution Methode zum Berechnen eines Werts aufruft. In diesem Beispiel ist nur das Ergebnis des ersten Aufrufs von TrySolution erforderlich, um abzuschließen.

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 {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.
'

Um den Wert aus dem ersten TransformBlock<TInput,TOutput> Objekt zu erhalten, das abgeschlossen ist, definiert dieses Beispiel die ReceiveFromAny(T) Methode. Die ReceiveFromAny(T) Methode akzeptiert ein Array von ISourceBlock<TOutput> Objekten und verknüpft jedes dieser Objekte mit einem WriteOnceBlock<T> Objekt. Wenn Sie die LinkTo Methode verwenden, um einen Quelldatenflussblock mit einem Zielblock zu verknüpfen, verteilt die Quelle Nachrichten an das Ziel, sobald Daten verfügbar werden. Da die WriteOnceBlock<T> Klasse nur die erste Nachricht akzeptiert, die sie anbietet, erzeugt die ReceiveFromAny(T) Methode ihr Ergebnis durch Aufrufen der Receive Methode. Dies erzeugt die erste Nachricht, die dem WriteOnceBlock<T> Objekt angeboten wird. Die LinkTo Methode verfügt über eine überladene Version, die ein DataflowLinkOptions Objekt mit einer MaxMessages Eigenschaft verwendet, die, wenn sie auf 1 gesetzt ist, den Quellblock anweist, die Verknüpfung vom Ziel aufzuheben, nachdem das Ziel eine Nachricht von der Quelle erhält. Es ist wichtig, dass das WriteOnceBlock<T> Objekt die Verknüpfung von seinen Quellen aufheben kann, da die Beziehung zwischen dem Array von Quellen und dem WriteOnceBlock<T> Objekt nicht mehr erforderlich ist, nachdem das WriteOnceBlock<T> Objekt eine Nachricht empfängt.

Die verbleibenden Aufrufe von TrySolution sollen beendet werden, nachdem einer von ihnen einen Wert berechnet. Daher nimmt die TrySolution-Methode ein CancellationToken-Objekt entgegen, das storniert wird, nachdem der Aufruf von ReceiveFromAny(T) zurückkehrt. Die Methode SpinUntil gibt zurück, wenn dieses Objekt CancellationToken storniert wird.

Siehe auch

• Datenfluss