?

本文主要演示使用TPL 数据流库从数据流块（dataflow block）读写消息。

提供了同步方法和异步方法。

主要使用BufferBlock，其既能作为message source，有能作为message target。

1. Install-Package Microsoft.Tpl.Dataflow
2. ?
3. using System.Threading.Tasks.Dataflow;

?

1.同步方法读写消息

?

1.1 分别使用Post和Receive方法进行读写。

1. public void SynchronouslyPostAndReceive()
2. {
3. ????var bufferBlock = new BufferBlock<int>();
4. ?
5. ????for (int i = 0; i < 3; i++)
6. ????{
7. ????????bufferBlock.Post(i);
8. ????}
9. ?
10. ????for (int i = 0; i < 3; i++)
11. ????{
12. ????????Console.WriteLine(bufferBlock.Receive());
13. ????}
14. }

输出：

0

1

2

解释：

?

从Post方法可以看出，此时BufferBlock扮演着ITargetBlock的角色；

同理，Receive方法调用时，BufferBlock扮演着ISourceBlock的角色；

这也就证明了BufferBlock同时用作Target和Source。已可以看出，同步的时候，BufferBlock实际上是FIFO。

?

1.2 使用TryReceive来读取

1. public void SynchronouslyPostAndReceive()
2. {
3. ????var bufferBlock = new BufferBlock<int>();
4. ?
5. ????for (int i = 0; i < 3; i++)
6. ????{
7. ????????bufferBlock.Post(i);
8. ????}
9. ?
10. ????//for (int i = 0; i < 3; i++)
11. ????//{
12. ????// Console.WriteLine(bufferBlock.Receive());
13. ????//}
14. ?
15. ????int value;
16. ????while (bufferBlock.TryReceive(out value))
17. ????{
18. ????????Console.WriteLine(value);
19. ????}
20. }

输出：

0

1

2

1.3并发读写

使用Task，将读写放在不同的线程。

当Receive，且无数据时，则会阻塞。

1. public void ConcurrentlyPostAndReceive()
2. {
3. ????var bufferBlock = new BufferBlock<int>();
4. ????var post01 = Task.Run(() =>
5. ????{
6. ????????bufferBlock.Post(1);
7. ????????bufferBlock.Post(2);
8. ????????bufferBlock.Post(3);
9. ????????bufferBlock.Post(4);
10. ????});
11. ?
12. ????var receive = Task.Run(() =>
13. ????{
14. ????????for (int i = 0; i < 6; i++)
15. ????????{
16. ????????????Console.WriteLine(bufferBlock.Receive());
17. ????????}
18. ????});
19. ?
20. ????var post02 = Task.Run(() =>
21. ????{
22. ????????bufferBlock.Post(5);
23. ????????bufferBlock.Post(6);
24. ????});
25. ?
26. ????Task.WaitAll(post01, receive, post02);
27. }

输出：

5

1

2

3

4

6

2.异步方式读写消息

?

主要是TAP的一种套用。使用async 和await来实现。

?

1. static async Task AsyncSendReceive(BufferBlock<int> bufferBlock)
2. {
3. ????// Post more messages to the block asynchronously.
4. ????for (int i = 0; i < 3; i++)
5. ????{
6. ????????await bufferBlock.SendAsync(i);
7. ????}
8. ?
9. ????// Asynchronously receive the messages back from the block.
10. ????for (int i = 0; i < 3; i++)
11. ????{
12. ????????Console.WriteLine(await bufferBlock.ReceiveAsync());
13. ????}
14. ?
15. }
16. ?
17. public void Start()
18. {
19. ????AsyncSendReceive(new BufferBlock<int>()).Wait();
20. }

HowTo:使用数据流读写消息

原文：http://www.cnblogs.com/pengzhen/p/4837434.html