Procedimiento Especificación del grado de paralelismo en un bloque de flujos de datosHow to: Specify the Degree of Parallelism in a Dataflow Block

En este documento se describe cómo establecer la propiedad ExecutionDataflowBlockOptions.MaxDegreeOfParallelism para que un bloque de flujo de datos de ejecución pueda procesar más de un mensaje al mismo tiempo.This document describes how to set the ExecutionDataflowBlockOptions.MaxDegreeOfParallelism property to enable an execution dataflow block to process more than one message at a time. Esto resulta útil cuando tiene un bloque de flujo de datos que realiza un cálculo de ejecución prolongada y se puede beneficiar del procesamiento de mensajes en paralelo.Doing this is useful when you have a dataflow block that performs a long-running computation and can benefit from processing messages in parallel. En el ejemplo se usa la clase System.Threading.Tasks.Dataflow.ActionBlock<TInput> para llevar a cabo varias operaciones de flujo de datos simultáneamente; sin embargo, puede especificar el grado máximo de paralelismo en cualquiera de los tipos de bloque de ejecución predefinidos que la biblioteca de flujo de datos TPL proporciona, ActionBlock<TInput>, System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput> y System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput>.This example uses the System.Threading.Tasks.Dataflow.ActionBlock<TInput> class to perform multiple dataflow operations concurrently; however, you can specify the maximum degree of parallelism in any of the predefined execution block types that the TPL Dataflow Library provides, ActionBlock<TInput>, System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput>, and System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput>.

Nota

La biblioteca de flujos de datos TPL (el espacio de nombres System.Threading.Tasks.Dataflow) no se distribuye con .NET.The TPL Dataflow Library (the System.Threading.Tasks.Dataflow namespace) is not distributed with .NET. Para instalar el espacio de nombres System.Threading.Tasks.Dataflow en Visual Studio, abra el proyecto, seleccione Administrar paquetes NuGet en el menú Proyecto y busque en línea el paquete System.Threading.Tasks.Dataflow.To install the System.Threading.Tasks.Dataflow namespace in Visual Studio, open your project, choose Manage NuGet Packages from the Project menu, and search online for the System.Threading.Tasks.Dataflow package. Como alternativa, para realizar la instalación con la CLI de .Net Core, ejecute dotnet add package System.Threading.Tasks.Dataflow.Alternatively, to install it using the .Net Core CLI, run dotnet add package System.Threading.Tasks.Dataflow.

EjemploExample

En el ejemplo siguiente se realizan dos cálculos de flujo de datos y se imprime el tiempo transcurrido que se necesita para cada cálculo.The following example performs two dataflow computations and prints the elapsed time that is required for each computation. El primer cálculo especifica un grado máximo de paralelismo de 1, que es el valor predeterminado.The first computation specifies a maximum degree of parallelism of 1, which is the default. Un grado máximo de paralelismo de 1 hace que el bloque de flujo de datos procese los mensajes en serie.A maximum degree of parallelism of 1 causes the dataflow block to process messages serially. El segundo cálculo es similar al primero, excepto que especifica un grado máximo de paralelismo igual al número de procesadores disponibles.The second computation resembles the first, except that it specifies a maximum degree of parallelism that is equal to the number of available processors. Esto permite que el bloque de flujo de datos realice varias operaciones en paralelo.This enables the dataflow block to perform multiple operations in parallel.

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

Programación sólidaRobust Programming

De forma predeterminada, cada bloque de flujo de datos predefinido propaga los mensajes en el orden con que se reciben.By default, each predefined dataflow block propagates out messages in the order in which the messages are received. Aunque cuando se especifica un grado máximo de paralelismo mayor que 1 se procesan simultáneamente varios mensajes, se siguen propagando en el orden con que se reciben.Although multiple messages are processed simultaneously when you specify a maximum degree of parallelism that is greater than 1, they are still propagated out in the order in which they are received.

Dado que la propiedad MaxDegreeOfParallelism representa el grado máximo de paralelismo, el bloque de flujo de datos puede ejecutarse con un menor grado de paralelismo que el especificado.Because the MaxDegreeOfParallelism property represents the maximum degree of parallelism, the dataflow block might execute with a lesser degree of parallelism than you specify. El bloque de flujo de datos puede usar un grado de paralelismo menor para cumplir los requisitos funcionales o porque hay una falta de recursos del sistema.The dataflow block can use a lesser degree of parallelism to meet its functional requirements or to account for a lack of available system resources. Un bloque de flujo de datos nunca elige un grado de paralelismo mayor que el especificado.A dataflow block never chooses a greater degree of parallelism than you specify.

Vea tambiénSee also