方法: プロデューサーとコンシューマーのデータフロー パターンを実装する

この記事では、TPL データフロー ライブラリを使用してプロデューサー/コンシューマー パターンを実装する方法について説明します。 このパターンでは、 プロデューサー はメッセージ ブロックにメッセージを送信し、 コンシューマー はそのブロックからメッセージを読み取ります。

Example

次の例は、データフローを使用する基本的なプロデューサー/コンシューマー モデルを示しています。 Produce メソッドは、ランダムなデータ バイトを含む配列をSystem.Threading.Tasks.Dataflow.ITargetBlock<TInput> オブジェクトに書き込み、Consume メソッドはSystem.Threading.Tasks.Dataflow.ISourceBlock<TOutput> オブジェクトからバイトを読み取ります。 派生型ではなく、 ISourceBlock<TOutput> インターフェイスと ITargetBlock<TInput> インターフェイスに対して動作することで、さまざまなデータフロー ブロック型に対して動作できる再利用可能なコードを記述できます。 この例では、 BufferBlock<T> クラスを使用します。 BufferBlock<T> クラスはソース ブロックとターゲット ブロックの両方として機能するため、プロデューサーとコンシューマーは共有オブジェクトを使用してデータを転送できます。

Produce メソッドは、ループ内のPost メソッドを呼び出して、ターゲット ブロックにデータを同期的に書き込みます。 Produce メソッドは、すべてのデータをターゲット ブロックに書き込んだ後、Complete メソッドを呼び出して、ブロックに追加のデータを使用できないように指定します。 Consume メソッドは、async 演算子と await 演算子 (Visual Basic では Async および Await) を使用して、ISourceBlock<TOutput> オブジェクトから受信したバイトの合計数を非同期的に計算します。 非同期的に動作するために、 Consume メソッドは OutputAvailableAsync メソッドを呼び出して、ソース ブロックに使用可能なデータがあり、ソース ブロックに追加のデータが使用できない場合に通知を受け取ります。

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.

堅牢なプログラミング

前の例では、1 つのコンシューマーのみを使用してソース データを処理します。 アプリケーションに複数のコンシューマーがある場合は、次の例に示すように、 TryReceive メソッドを使用してソース ブロックからデータを読み取ります。

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を返します。 複数のコンシューマーが同時にソース ブロックにアクセスする必要がある場合、このメカニズムにより、 OutputAvailableAsyncの呼び出し後もデータが使用可能であることが保証されます。

こちらも参照ください

• Dataflow