本文介绍如何使用任务并行库（TPL）数据流库向数据流块写入和读取消息。 TPL 数据流库提供同步和异步方法，用于向数据流块写入和读取消息。 本文介绍如何使用该 System.Threading.Tasks.Dataflow.BufferBlock<T> 类。 类 BufferBlock<T> 缓冲消息并充当消息源和消息目标。
注释
TPL 数据流库（System.Threading.Tasks.Dataflow命名空间）包含在 .NET 6 及更高版本中。 对于 .NET Framework 和 .NET Standard 项目，需要安装 📦 System.Threading.Tasks.Dataflow NuGet 包。
同步写入和读取
以下示例使用Post方法写入BufferBlock<T>数据流块，并使用Receive方法从同一对象中读取。
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
还可以使用 TryReceive 该方法从数据流块中读取数据，如以下示例所示。 此方法 TryReceive 不会阻止当前线程，在偶尔轮询数据时非常有用。
// 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
由于Post 方法以同步方式运行，在上一个示例中的BufferBlock<T> 对象在第二个循环读取数据之前接收了所有数据。 以下示例通过使用 Task.WhenAll(Task[]) 并发读取消息块和写入消息块来扩展第一个示例。 由于 WhenAll 等待所有同时执行的异步作，因此值不会以任何特定顺序写入 BufferBlock<T> 对象。
// 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
异步写入和读取
以下示例使用SendAsync方法异步写入一个BufferBlock<T>对象，并使用ReceiveAsync方法从同一对象异步读取。 此示例使用 async 和 await 运算符（ Async 和 Await in Visual Basic） 以异步方式向目标块发送和读取数据。 当必须启用数据流块以推迟消息时，此方法 SendAsync 非常有用。 当想要在数据可用时对数据采取行动时，此方法 ReceiveAsync 非常有用。 有关消息如何在消息块之间传播的详细信息，请参阅 “数据流中的消息传递”部分。
// 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
完整示例
以下示例演示本文的所有代码。
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
后续步骤
此示例演示如何直接从消息块读取和写入消息块。 您还可以连接数据流块以形成管道，这些管道是数据流块的线性序列，或网络，即数据流块的图形结构。 在管道或网络中，源在数据可用时以异步方式将数据传播到目标。 有关创建基本数据流管道的示例，请参阅 演练：创建数据流管道。 有关创建更复杂的数据流网络的示例，请参阅 Walkthrough：在 Windows 窗体 Application 中使用数据流。
