Практическое руководство. Реализация шаблона потока данных "производитель-потребитель"How to: Implement a producer-consumer dataflow pattern

Из этой статьи вы узнаете, как использовать библиотеки потоков данных TPL для реализации шаблона "производитель-получатель".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.

Примечание

Библиотека потоков данных TPL (пространство имен System.Threading.Tasks.Dataflow) не поставляется с .NET.The TPL Dataflow Library (the System.Threading.Tasks.Dataflow namespace) is not distributed with .NET. Чтобы установить пространство имен System.Threading.Tasks.Dataflow в Visual Studio, откройте проект, выберите Управление пакетами NuGet в меню Проект и выполните поиск пакета 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. Вы также можете установить его, выполнив в .NET Core CLI команду dotnet add package System.Threading.Tasks.Dataflow.Alternatively, to install it using the .NET Core CLI, run dotnet add package System.Threading.Tasks.Dataflow.

ПримерExample

В следующем примере показана базовая модель "производитель-получатель", которая использует поток данных.The following example demonstrates a basic producer- consumer model that uses dataflow. Метод Produce записывает массивы, содержащие случайные байты данных, в объект System.Threading.Tasks.Dataflow.ITargetBlock<TInput>, а метод Consume выполняет чтение байтов из объекта System.Threading.Tasks.Dataflow.ISourceBlock<TOutput>.The Produce method writes arrays that contain random bytes of data to a System.Threading.Tasks.Dataflow.ITargetBlock<TInput> object and the Consume method reads bytes from a System.Threading.Tasks.Dataflow.ISourceBlock<TOutput> object. Используя интерфейсы ISourceBlock<TOutput> и ITargetBlock<TInput> вместо их производных типов, можно создавать пригодный для повторного использования код, который может работать с различными типами блоков потока данных.By acting on the ISourceBlock<TOutput> and ITargetBlock<TInput> interfaces, instead of their derived types, you can write reusable code that can act on a variety of dataflow block types. В этом примере используется класс BufferBlock<T>.This example uses the BufferBlock<T> class. Поскольку класс BufferBlock<T> действует и как блок источника, и как целевой блок, потребитель и производитель могут использовать общий объект для передачи данных.Because the BufferBlock<T> 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.

Метод Produce вызывает метод Post в цикле для синхронной записи данных в целевой блок.The Produce method calls the Post method in a loop to synchronously write data to the target block. После того, как метод Produce записывает все данные в целевой блок, он вызывает метод Complete, чтобы указать, что у этого блока никогда не будет дополнительных доступных данных.After the Produce method writes all data to the target block, it calls the Complete method to indicate that the block will never have additional data available. Метод Consume использует операторы async и await (Async и Await в Visual Basic) для асинхронного вычисления общего числа байтов, полученных от объекта ISourceBlock<TOutput>.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 ISourceBlock<TOutput> object. Для асинхронной работы метод Consume вызывает метод OutputAvailableAsync, чтобы получать уведомления, если блок источника получит доступные данные и если у блока источника никогда не будет дополнительных доступных данных.To act asynchronously, the Consume method calls the OutputAvailableAsync method to receive a notification when the source block has data available and when the source block will never have additional data available.

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

ОтказоустойчивостьRobust programming

В предыдущем примере используется только один получатель для обработки исходных данных.The preceding example uses just one consumer to process the source data. При наличии нескольких получателей в приложении следует использовать метод TryReceive для чтения данных из блока источника, как показано в следующем примере.If you have multiple consumers in your application, use the TryReceive method to read data from the source block, as shown in the following example.

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;
}
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

Метод TryReceive возвращает False, когда нет доступных данных.The TryReceive method returns False when no data is available. Когда несколько потребителей должны использовать блок источника параллельно, этот механизм гарантирует, что данные все еще будут доступны после вызова OutputAvailableAsync.When multiple consumers must access the source block concurrently, this mechanism guarantees that data is still available after the call to OutputAvailableAsync.

См. такжеSee also