How to: Implement a producer-consumer dataflow pattern

In this article, you'll learn how to use the TPL dataflow library to implement a producer-consumer pattern. In this pattern, the producer sends messages to a message block, and the consumer reads messages from that block.

Example

The following example demonstrates a basic producer-consumer model that uses dataflow. The Produce method writes arrays that contain random bytes of data to a System.Threading.Tasks.Dataflow.ITargetBlock object and the Consume method reads bytes from a System.Threading.Tasks.Dataflow.ISourceBlock object. By acting on the System.Threading.Tasks.Dataflow.ISourceBlock and System.Threading.Tasks.Dataflow.ITargetBlock interfaces, instead of their derived types, you can write reusable code that can act on a variety of dataflow block types. This example uses the System.Threading.Tasks.Dataflow.BufferBlock class. Because the System.Threading.Tasks.Dataflow.BufferBlock class acts as both a source block and as a target block, the producer and the consumer can use a shared object to transfer data.

The Produce method calls the System.Threading.Tasks.Dataflow.DataflowBlock.Post method in a loop to synchronously write data to the target block. After the Produce method writes all data to the target block, it calls the System.Threading.Tasks.Dataflow.IDataflowBlock.Complete method to indicate that the block will never have additional data available. The Consume method uses the async and await operators (Async and Await in Visual Basic) to asynchronously compute the total number of bytes that are received from the System.Threading.Tasks.Dataflow.ISourceBlock object. To act asynchronously, the Consume method calls the System.Threading.Tasks.Dataflow.DataflowBlock.OutputAvailableAsync method to receive a notification when the source block has data available and when the source block will never have additional data available.

sample · csharpsamples/snippets/csharp/VS_Snippets_Misc/tpldataflow_producerconsumer/cs/dataflowproducerconsumer.cs1

csharp

// <snippet1>
using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

class DataflowProducerConsumer
{
    static void Produce(ITargetBlock<byte[]> target)
    {
        var rand = new Random();

        for (int i = 0; i < 100; ++ i)
        {
            var buffer = new byte[1024];
            rand.NextBytes(buffer);
            target.Post(buffer);
        }

        target.Complete();
    }

    static async Task<int> ConsumeAsync(ISourceBlock<byte[]> source)
    {
        int bytesProcessed = 0;

        while (await source.OutputAvailableAsync())
        {
            byte[] data = await source.ReceiveAsync();
            bytesProcessed += data.Length;
        }

        return bytesProcessed;
    }

    static async Task Main()
    {
        var buffer = new BufferBlock<byte[]>();
        var consumerTask = ConsumeAsync(buffer);
        Produce(buffer);

        var bytesProcessed = await consumerTask;

        Console.WriteLine($"Processed {bytesProcessed:#,#} bytes.");
    }
}

// Sample  output:
//     Processed 102,400 bytes.
// </snippet1>

class Program2
{
    // <snippet2>
    static async Task<int> ConsumeAsync(IReceivableSourceBlock<byte[]> source)
    {
        int bytesProcessed = 0;
        while (await source.OutputAvailableAsync())
        {
            while (source.TryReceive(out byte[] data))
            {
                bytesProcessed += data.Length;
            }
        }
        return bytesProcessed;
    }
    // </snippet2>
}

sample · vbsamples/snippets/visualbasic/VS_Snippets_Misc/tpldataflow_producerconsumer/vb/dataflowproducerconsumer.vb1

' <snippet1>
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow

Friend Class DataflowProducerConsumer
    Private Shared Sub Produce(ByVal target As ITargetBlock(Of Byte()))
        Dim rand As New Random()

        For i As Integer = 0 To 99
            Dim buffer(1023) As Byte
            rand.NextBytes(buffer)
            target.Post(buffer)
        Next i

        target.Complete()
    End Sub

    Private Shared Async Function ConsumeAsync(
        ByVal source As ISourceBlock(Of Byte())) As Task(Of Integer)
        Dim bytesProcessed As Integer = 0

        Do While Await source.OutputAvailableAsync()
            Dim data() As Byte = Await source.ReceiveAsync()
            bytesProcessed += data.Length
        Loop

        Return bytesProcessed
    End Function

    Shared Sub Main()
        Dim buffer = New BufferBlock(Of Byte())()
        Dim consumer = ConsumeAsync(buffer)
        Produce(buffer)

        Dim result = consumer.GetAwaiter().GetResult()

        Console.WriteLine($"Processed {result:#,#} bytes.")
    End Sub
End Class

' Sample output:
'     Processed 102,400 bytes.
' </snippet1>

Friend Class Program2
    ' <snippet2>
    Private Shared Async Function ConsumeAsync(
        ByVal source As IReceivableSourceBlock(Of Byte())) As Task(Of Integer)
        Dim bytesProcessed As Integer = 0
        
        Do While Await source.OutputAvailableAsync()
            Dim data() As Byte
            Do While source.TryReceive(data)
                bytesProcessed += data.Length
            Loop
        Loop

        Return bytesProcessed
    End Function
    ' </snippet2>
End Class

Robust programming

The preceding example uses just one consumer to process the source data. If you have multiple consumers in your application, use the System.Threading.Tasks.Dataflow.IReceivableSourceBlock.TryReceive method to read data from the source block, as shown in the following example.

sample · csharpsamples/snippets/csharp/VS_Snippets_Misc/tpldataflow_producerconsumer/cs/dataflowproducerconsumer.cs2

csharp

// <snippet1>
using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

class DataflowProducerConsumer
{
    static void Produce(ITargetBlock<byte[]> target)
    {
        var rand = new Random();

        for (int i = 0; i < 100; ++ i)
        {
            var buffer = new byte[1024];
            rand.NextBytes(buffer);
            target.Post(buffer);
        }

        target.Complete();
    }

    static async Task<int> ConsumeAsync(ISourceBlock<byte[]> source)
    {
        int bytesProcessed = 0;

        while (await source.OutputAvailableAsync())
        {
            byte[] data = await source.ReceiveAsync();
            bytesProcessed += data.Length;
        }

        return bytesProcessed;
    }

    static async Task Main()
    {
        var buffer = new BufferBlock<byte[]>();
        var consumerTask = ConsumeAsync(buffer);
        Produce(buffer);

        var bytesProcessed = await consumerTask;

        Console.WriteLine($"Processed {bytesProcessed:#,#} bytes.");
    }
}

// Sample  output:
//     Processed 102,400 bytes.
// </snippet1>

class Program2
{
    // <snippet2>
    static async Task<int> ConsumeAsync(IReceivableSourceBlock<byte[]> source)
    {
        int bytesProcessed = 0;
        while (await source.OutputAvailableAsync())
        {
            while (source.TryReceive(out byte[] data))
            {
                bytesProcessed += data.Length;
            }
        }
        return bytesProcessed;
    }
    // </snippet2>
}

sample · vbsamples/snippets/visualbasic/VS_Snippets_Misc/tpldataflow_producerconsumer/vb/dataflowproducerconsumer.vb2

' <snippet1>
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow

Friend Class DataflowProducerConsumer
    Private Shared Sub Produce(ByVal target As ITargetBlock(Of Byte()))
        Dim rand As New Random()

        For i As Integer = 0 To 99
            Dim buffer(1023) As Byte
            rand.NextBytes(buffer)
            target.Post(buffer)
        Next i

        target.Complete()
    End Sub

    Private Shared Async Function ConsumeAsync(
        ByVal source As ISourceBlock(Of Byte())) As Task(Of Integer)
        Dim bytesProcessed As Integer = 0

        Do While Await source.OutputAvailableAsync()
            Dim data() As Byte = Await source.ReceiveAsync()
            bytesProcessed += data.Length
        Loop

        Return bytesProcessed
    End Function

    Shared Sub Main()
        Dim buffer = New BufferBlock(Of Byte())()
        Dim consumer = ConsumeAsync(buffer)
        Produce(buffer)

        Dim result = consumer.GetAwaiter().GetResult()

        Console.WriteLine($"Processed {result:#,#} bytes.")
    End Sub
End Class

' Sample output:
'     Processed 102,400 bytes.
' </snippet1>

Friend Class Program2
    ' <snippet2>
    Private Shared Async Function ConsumeAsync(
        ByVal source As IReceivableSourceBlock(Of Byte())) As Task(Of Integer)
        Dim bytesProcessed As Integer = 0
        
        Do While Await source.OutputAvailableAsync()
            Dim data() As Byte
            Do While source.TryReceive(data)
                bytesProcessed += data.Length
            Loop
        Loop

        Return bytesProcessed
    End Function
    ' </snippet2>
End Class

The System.Threading.Tasks.Dataflow.IReceivableSourceBlock.TryReceive method returns False when no data is available. When multiple consumers must access the source block concurrently, this mechanism guarantees that data is still available after the call to System.Threading.Tasks.Dataflow.DataflowBlock.OutputAvailableAsync.

#Example

#Robust programming

#See also

Example

Robust programming

See also