Procedura: Scrivere e leggere messaggi da un blocco di flusso di dati

Questo articolo descrive come usare la libreria di flussi di dati TPL (Task Parallel Library) per scrivere e leggere messaggi da un blocco di flussi di dati. La libreria di flussi di dati TPL fornisce metodi sincroni e asincroni per scrivere messaggi e leggere messaggi da un blocco di flussi di dati. Questo articolo illustra come usare la System.Threading.Tasks.Dataflow.BufferBlock<T> classe . La BufferBlock<T> classe memorizza nel buffer i messaggi e si comporta sia come origine del messaggio che come destinazione del messaggio.

Scrittura e lettura in modo sincrono

Nell'esempio seguente viene utilizzato il Post metodo per scrivere in un BufferBlock<T> blocco di flussi di dati e il Receive metodo da leggere dallo stesso oggetto.

var bufferBlock = new BufferBlock<int>();

// Post several messages to the block.
for (int i = 0; i < 3; i++)
{
    bufferBlock.Post(i);
}

// Receive the messages back from the block.
for (int i = 0; i < 3; i++)
{
    Console.WriteLine(bufferBlock.Receive());
}

// Output:
//   0
//   1
//   2

Dim bufferBlock = New BufferBlock(Of Integer)()

' Post several messages to the block.
For i As Integer = 0 To 2
    bufferBlock.Post(i)
Next i

' Receive the messages back from the block.
For i As Integer = 0 To 2
    Console.WriteLine(bufferBlock.Receive())
Next i

' Output:
'   0
'   1
'   2

È anche possibile usare il TryReceive metodo per leggere da un blocco di flussi di dati, come illustrato nell'esempio seguente. Il metodo TryReceive non blocca il thread corrente ed è utile quando si effettua un polling dei dati occasionalmente.

// Post more messages to the block.
for (int i = 0; i < 3; i++)
{
    bufferBlock.Post(i);
}

// Receive the messages back from the block.
while (bufferBlock.TryReceive(out int value))
{
    Console.WriteLine(value);
}

// Output:
//   0
//   1
//   2

' Post more messages to the block.
For i As Integer = 0 To 2
    bufferBlock.Post(i)
Next i

' Receive the messages back from the block.
Dim value As Integer
Do While bufferBlock.TryReceive(value)
    Console.WriteLine(value)
Loop

' Output:
'   0
'   1
'   2

Poiché il Post metodo agisce in modo sincrono, l'oggetto BufferBlock<T> negli esempi precedenti riceve tutti i dati prima che il secondo ciclo legga i dati. Nell'esempio seguente viene esteso il primo esempio usando Task.WhenAll(Task[]) per leggere e scrivere simultaneamente nel blocco di messaggi. Poiché WhenAll attende tutte le operazioni asincrone in esecuzione contemporaneamente, i valori non vengono scritti nell'oggetto BufferBlock<T> in un ordine specifico.

// Write to and read from the message block concurrently.
var post01 = Task.Run(() =>
{
    bufferBlock.Post(0);
    bufferBlock.Post(1);
});
var receive = Task.Run(() =>
{
    for (int i = 0; i < 3; i++)
    {
        Console.WriteLine(bufferBlock.Receive());
    }
});
var post2 = Task.Run(() =>
{
    bufferBlock.Post(2);
});

await Task.WhenAll(post01, receive, post2);

// Output:
//   0
//   1
//   2

' Write to and read from the message block concurrently.
Dim post01 = Task.Run(Sub()
                          bufferBlock.Post(0)
                          bufferBlock.Post(1)
                      End Sub)
Dim receive = Task.Run(Sub()
                           For i As Integer = 0 To 2
                               Console.WriteLine(bufferBlock.Receive())
                           Next i
                       End Sub)
Dim post2 = Task.Run(Sub() bufferBlock.Post(2))
Task.WaitAll(post01, receive, post2)

' Output:
'   0
'   1
'   2

Scrittura e lettura in modo asincrono

Nell'esempio seguente viene utilizzato il SendAsync metodo per scrivere in modo asincrono in un BufferBlock<T> oggetto e il ReceiveAsync metodo per leggere in modo asincrono dallo stesso oggetto. In questo esempio vengono usati gli operatori async e await (Async e Await in Visual Basic) per inviare dati in modo asincrono e leggere i dati dal blocco di destinazione. Il SendAsync metodo è utile quando è necessario abilitare un blocco di flussi di dati per posticipare i messaggi. Il ReceiveAsync metodo è utile quando si desidera agire sui dati quando questi dati diventano disponibili. Per altre informazioni sulla propagazione dei messaggi tra blocchi di messaggi, vedere la sezione Passaggio di messaggi nel flusso di dati.

// Post more messages to the block asynchronously.
for (int i = 0; i < 3; i++)
{
    await bufferBlock.SendAsync(i);
}

// Asynchronously receive the messages back from the block.
for (int i = 0; i < 3; i++)
{
    Console.WriteLine(await bufferBlock.ReceiveAsync());
}

// Output:
//   0
//   1
//   2

' Post more messages to the block asynchronously.
For i As Integer = 0 To 2
    await bufferBlock.SendAsync(i)
Next i

' Asynchronously receive the messages back from the block.
For i As Integer = 0 To 2
    Console.WriteLine(await bufferBlock.ReceiveAsync())
Next i

' Output:
'   0
'   1
'   2

Esempio completo

Nell'esempio seguente viene illustrato tutto il codice per questo articolo.

using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

// Demonstrates a how to write to and read from a dataflow block.
class DataflowReadWrite
{
    // Demonstrates asynchronous dataflow operations.
    static async Task AsyncSendReceive(BufferBlock<int> bufferBlock)
    {
        // Post more messages to the block asynchronously.
        for (int i = 0; i < 3; i++)
        {
            await bufferBlock.SendAsync(i);
        }

        // Asynchronously receive the messages back from the block.
        for (int i = 0; i < 3; i++)
        {
            Console.WriteLine(await bufferBlock.ReceiveAsync());
        }

        // Output:
        //   0
        //   1
        //   2
    }

    static async Task Main()
    {
        var bufferBlock = new BufferBlock<int>();

        // Post several messages to the block.
        for (int i = 0; i < 3; i++)
        {
            bufferBlock.Post(i);
        }

        // Receive the messages back from the block.
        for (int i = 0; i < 3; i++)
        {
            Console.WriteLine(bufferBlock.Receive());
        }

        // Output:
        //   0
        //   1
        //   2

        // Post more messages to the block.
        for (int i = 0; i < 3; i++)
        {
            bufferBlock.Post(i);
        }

        // Receive the messages back from the block.
        while (bufferBlock.TryReceive(out int value))
        {
            Console.WriteLine(value);
        }

        // Output:
        //   0
        //   1
        //   2

        // Write to and read from the message block concurrently.
        var post01 = Task.Run(() =>
        {
            bufferBlock.Post(0);
            bufferBlock.Post(1);
        });
        var receive = Task.Run(() =>
        {
            for (int i = 0; i < 3; i++)
            {
                Console.WriteLine(bufferBlock.Receive());
            }
        });
        var post2 = Task.Run(() =>
        {
            bufferBlock.Post(2);
        });

        await Task.WhenAll(post01, receive, post2);

        // Output:
        //   0
        //   1
        //   2

        // Demonstrate asynchronous dataflow operations.
        await AsyncSendReceive(bufferBlock);
    }
}

Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow

' Demonstrates a how to write to and read from a dataflow block.
Friend Class DataflowReadWrite
    ' Demonstrates asynchronous dataflow operations.
    Private Shared async Function AsyncSendReceive(ByVal bufferBlock As BufferBlock(Of Integer)) As Task
        ' Post more messages to the block asynchronously.
        For i As Integer = 0 To 2
            await bufferBlock.SendAsync(i)
        Next i

        ' Asynchronously receive the messages back from the block.
        For i As Integer = 0 To 2
            Console.WriteLine(await bufferBlock.ReceiveAsync())
        Next i

        ' Output:
        '   0
        '   1
        '   2
    End Function

    Shared Sub Main(ByVal args() As String)
        Dim bufferBlock = New BufferBlock(Of Integer)()

        ' Post several messages to the block.
        For i As Integer = 0 To 2
            bufferBlock.Post(i)
        Next i

        ' Receive the messages back from the block.
        For i As Integer = 0 To 2
            Console.WriteLine(bufferBlock.Receive())
        Next i

        ' Output:
        '   0
        '   1
        '   2

        ' Post more messages to the block.
        For i As Integer = 0 To 2
            bufferBlock.Post(i)
        Next i

        ' Receive the messages back from the block.
        Dim value As Integer
        Do While bufferBlock.TryReceive(value)
            Console.WriteLine(value)
        Loop

        ' Output:
        '   0
        '   1
        '   2

        ' Write to and read from the message block concurrently.
        Dim post01 = Task.Run(Sub()
                                  bufferBlock.Post(0)
                                  bufferBlock.Post(1)
                              End Sub)
        Dim receive = Task.Run(Sub()
                                   For i As Integer = 0 To 2
                                       Console.WriteLine(bufferBlock.Receive())
                                   Next i
                               End Sub)
        Dim post2 = Task.Run(Sub() bufferBlock.Post(2))
        Task.WaitAll(post01, receive, post2)

        ' Output:
        '   0
        '   1
        '   2

        ' Demonstrate asynchronous dataflow operations.
        AsyncSendReceive(bufferBlock).Wait()
    End Sub

End Class

Passaggi successivi

In questo esempio viene illustrato come leggere e scrivere direttamente in un blocco di messaggi. È anche possibile connettere blocchi di flussi di dati per formare pipeline, che sono sequenze lineari di blocchi di flussi di dati, o reti, che sono grafi di blocchi di flussi di dati. In una pipeline o in una rete le origini propagano in modo asincrono i dati alle destinazioni man mano che i dati diventano disponibili. Per un esempio che crea una pipeline di flusso di dati di base, vedere Procedura dettagliata: Creazione di una pipeline del flusso di dati. Per un esempio che crea una rete di flussi di dati più complessa, vedere Procedura dettagliata: Uso del flusso di dati in un'applicazione Windows Form.

Vedere anche

• Flusso di dati (Task Parallel Library)