Tutorial: Creación de una canalización de flujos de datosWalkthrough: Creating a Dataflow Pipeline

03/30/2017

Tiempo de lectura: 9 minutos

En este artículo

Aunque puede usar los métodos DataflowBlock.Receive, DataflowBlock.ReceiveAsync y DataflowBlock.TryReceive para recibir mensajes de los bloques de origen, también puede conectar los bloques de mensajes para formar una canalización de flujo de datos.Although you can use the DataflowBlock.Receive, DataflowBlock.ReceiveAsync, and DataflowBlock.TryReceive methods to receive messages from source blocks, you can also connect message blocks to form a dataflow pipeline.Una canalización de flujo datos es una serie de componentes, o bloques de flujo de datos, de los que cada uno realiza una tarea concreta que contribuye a lograr un objetivo mayor.A dataflow pipeline is a series of components, or dataflow blocks, each of which performs a specific task that contributes to a larger goal.Cada bloque de flujo de datos de una canalización de flujo de datos realiza un determinado trabajo cuando recibe un mensaje de otro bloque de flujo de datos.Every dataflow block in a dataflow pipeline performs work when it receives a message from another dataflow block.Se podría establecer una analogía de esto con una cadena de montaje en la fabricación de automóviles.An analogy to this is an assembly line for automobile manufacturing.Mientras cada vehículo pasa a través de la línea de montaje, una estación monta el bastidor, la siguiente instala el motor y así sucesivamente.As each vehicle passes through the assembly line, one station assembles the frame, the next one installs the engine, and so on.Dado que una cadena de montaje permite montar varios vehículos al mismo tiempo, proporciona un mejor rendimiento que montar de uno en uno los vehículos completos.Because an assembly line enables multiple vehicles to be assembled at the same time, it provides better throughput than assembling complete vehicles one at a time.

Crear los bloques de flujo de datos que participan en la canalización.Create the dataflow blocks that participate in the pipeline.

Conectar cada bloque de flujo de datos con el siguiente bloque de la canalización.Connect each dataflow block to the next block in the pipeline.Cada bloque recibe como entrada la salida del bloque anterior de la canalización.Each block receives as input the output of the previous block in the pipeline.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Threading.Tasks.Dataflow;
// Demonstrates how to create a basic dataflow pipeline.
// This program downloads the book "The Iliad of Homer" by Homer from the Web
// and finds all reversed words that appear in that book.
static class Program
{
static void Main()
{
}
}

Imports System.Net.Http
Imports System.Threading.Tasks.Dataflow
' Demonstrates how to create a basic dataflow pipeline.
' This program downloads the book "The Iliad of Homer" by Homer from the Web
' and finds all reversed words that appear in that book.
Module DataflowReversedWords
Sub Main()
End Sub
End Module

Creación de los bloques de flujo de datosCreating the Dataflow Blocks

Agregue el código siguiente al método Main para crear los bloques de flujo de datos que participan en la canalización.Add the following code to the Main method to create the dataflow blocks that participate in the pipeline.En la tabla siguiente se resume el rol de cada miembro de la canalización.The table that follows summarizes the role of each member of the pipeline.

Envío de datos a la canalizaciónPosting Data to the Pipeline

Agregue el código siguiente para publicar la dirección URL del libro La Ilíada de Homero en el encabezado de la canalización de flujo de datos.Add the following code to post the URL of the book The Iliad of Homer to the head of the dataflow pipeline.

// Process "The Iliad of Homer" by Homer.
downloadString.Post("http://www.gutenberg.org/files/6130/6130-0.txt");

' Process "The Iliad of Homer" by Homer.
downloadString.Post("http://www.gutenberg.org/files/6130/6130-0.txt")

En este ejemplo se usa DataflowBlock.Post para enviar datos de forma sincrónica al encabezado de la canalización.This example uses DataflowBlock.Post to synchronously send data to the head of the pipeline.Use el método DataflowBlock.SendAsync cuando deba enviar datos de forma asincrónica a un nodo de flujo de datos.Use the DataflowBlock.SendAsync method when you must asynchronously send data to a dataflow node.

Agregue el código siguiente para marcar el encabezado de la canalización como completado.Add the following code to mark the head of the pipeline as completed.El encabezado de la canalización propaga su finalización después de procesar todos los mensajes almacenados en búfer.The head of the pipeline propagates its completion after it processes all buffered messages.

// Mark the head of the pipeline as complete.
downloadString.Complete();

' Mark the head of the pipeline as complete.
downloadString.Complete()

En este ejemplo se envía una dirección URL a través de la canalización de flujo de datos para que se procese.This example sends one URL through the dataflow pipeline to be processed.Si envía más de una entrada a través de una canalización, llame al método IDataflowBlock.Complete después de enviar toda la entrada.If you send more than one input through a pipeline, call the IDataflowBlock.Complete method after you submit all the input.Puede omitir este paso si la aplicación no tiene ningún punto bien definido en el que los datos ya no están disponibles o la aplicación no tiene que esperar a que finalice la canalización.You can omit this step if your application has no well-defined point at which data is no longer available or the application does not have to wait for the pipeline to finish.

Espera para la finalización de la canalizaciónWaiting for the Pipeline to Finish

Agregue el código siguiente para esperar a que finalice la canalización.Add the following code to wait for the pipeline to finish.La operación global termina cuando finaliza la cola de la canalización.The overall operation is finished when the tail of the pipeline finishes.

// Wait for the last block in the pipeline to process all messages.
printReversedWords.Completion.Wait();

' Wait for the last block in the pipeline to process all messages.
printReversedWords.Completion.Wait()

Pasos siguientesNext Steps

En este ejemplo se envía una dirección URL para procesarla a través de la canalización de flujo de datos.This example sends one URL to process through the dataflow pipeline.Si envía más de un valor de entrada a través de una canalización, puede introducir un formulario de paralelismo en la aplicación que se parezca a cómo se moverían las piezas en una fábrica de automóviles.If you send more than one input value through a pipeline, you can introduce a form of parallelism into your application that resembles how parts might move through an automobile factory.Cuando el primer miembro de la canalización envía su resultado al segundo miembro, puede procesar otro elemento en paralelo mientras el segundo miembro procesa el primer resultado.When the first member of the pipeline sends its result to the second member, it can process another item in parallel as the second member processes the first result.

El paralelismo que se logra mediante el uso de canalizaciones de flujo de datos se conoce como paralelismo general porque normalmente consta de menos tareas y más grandes.The parallelism that is achieved by using dataflow pipelines is known as coarse-grained parallelism because it typically consists of fewer, larger tasks.También puede usar un paralelismo específico de tareas más pequeñas y breves en una canalización de flujo de datos.You can also use a more fine-grained parallelism of smaller, short-running tasks in a dataflow pipeline.En este ejemplo, el miembro findReversedWords de la canalización usa PLINQ para procesar en paralelo varios elementos de la lista de trabajo.In this example, the findReversedWords member of the pipeline uses PLINQ to process multiple items in the work list in parallel.El uso de paralelismo de grano fino en una canalización de grano grueso puede mejorar el rendimiento global.The use of fine-grained parallelism in a coarse-grained pipeline can improve overall throughput.

También puede conectar un bloque de flujo de datos de origen a varios bloques de destino para crear una red de flujo de datos.You can also connect a source dataflow block to multiple target blocks to create a dataflow network.La versión sobrecargada del método LinkTo toma un objeto Predicate<T> que define si el bloque de destino acepta cada mensaje según su valor.The overloaded version of the LinkTo method takes a Predicate<T> object that defines whether the target block accepts each message based on its value.La mayoría de los tipos de bloques de flujo de datos que actúan como orígenes ofrecen mensajes a todos los bloques de destino conectados, siguiendo el orden en que se conectaron, hasta que uno de los bloques acepta ese mensaje.Most dataflow block types that act as sources offer messages to all connected target blocks, in the order in which they were connected, until one of the blocks accepts that message.Mediante este mecanismo de filtrado, puede crear sistemas de bloques de flujo de datos conectados que dirigen determinados datos a través de una ruta de acceso y otros datos a través de otra ruta de acceso.By using this filtering mechanism, you can create systems of connected dataflow blocks that direct certain data through one path and other data through another path.Para ver un ejemplo que usa el filtrado para crear una red de flujo de datos, consulte Tutorial: Uso de flujos de datos en aplicaciones de Windows Forms.For an example that uses filtering to create a dataflow network, see Walkthrough: Using Dataflow in a Windows Forms Application.