Útmutató: A párhuzamosság fokának megadása adatfolyamblokkokban

Ez a dokumentum azt ismerteti, hogyan állíthatja be a ExecutionDataflowBlockOptions.MaxDegreeOfParallelism tulajdonságot, hogy egy végrehajtási adatfolyam-blokk egyszerre több üzenetet is feldolgozhasson. Ez akkor hasznos, ha egy adatfolyamblokk hosszú ideig futó számítást végez, és kihasználhatja az üzenetek párhuzamos feldolgozását. Ez a példa a System.Threading.Tasks.Dataflow.ActionBlock<TInput> osztályt használja több adatfolyamművelet egyidejű végrehajtására; azonban megadhatja a párhuzamosság maximális fokát a TPL adatfolyamtár által biztosított előre definiált végrehajtási blokktípusokban, ActionBlock<TInput>, System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput>és System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput>.

Megjegyzés

A TPL-adatfolyamtár (a System.Threading.Tasks.Dataflow névtér) nincs elosztva a .NET-tel. Ha telepíteni szeretné a System.Threading.Tasks.Dataflow névteret a Visual Studióban, nyissa meg a projektet, válassza NuGet-csomagok kezelése a Project menüjében, és keressen online a System.Threading.Tasks.Dataflow csomagra. Másik lehetőségként a .NET Core CLI használatával történő telepítéséhez futtassa a dotnet add package System.Threading.Tasks.Dataflow.

Példa

Az alábbi példa két adatfolyam-számítást végez, és kinyomtatja az egyes számításokhoz szükséges eltelt időt. Az első számítás az 1 maximális párhuzamossági fokát adja meg, ez az alapértelmezett érték. Az 1 maximális párhuzamossági foka miatt az adatfolyam-blokk sorosan dolgozza fel az üzeneteket. A második számítás hasonlít az elsőre, azzal a kivétellel, hogy a párhuzamosság maximális fokát határozza meg, amely egyenlő a rendelkezésre álló processzorok számával. Ez lehetővé teszi, hogy az adatfolyam-blokk párhuzamosan több műveletet is végrehajtson.

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

Robusztus programozás

Alapértelmezés szerint minden előre definiált adatfolyamblokk az üzenetek fogadási sorrendjében propagálja az üzeneteket. Bár a rendszer egyszerre több üzenetet is feldolgoz, amikor 1-nél nagyobb párhuzamossági fokot ad meg, azok továbbra is a beérkezés sorrendjében lesznek propagálva.

Mivel a MaxDegreeOfParallelism tulajdonság a párhuzamosság maximális fokát jelöli, az adatfolyamblokk a megadottnál kisebb mértékű párhuzamossággal is végrehajtható. Az adatfolyam-blokk kisebb fokú párhuzamosságot használhat a funkcionális követelmények teljesítéséhez vagy a rendelkezésre álló rendszererőforrások hiányának figyelembevétele érdekében. Az adatfolyamblokkok soha nem választják ki a megadottnál nagyobb fokú párhuzamosságot.

Lásd még

• adatfolyam