How to: 执行Action当收到数据时

?

本文旨在演示ActionBlock的使用。

大致流程：

输入路径——读取字节——计算——传输到打印

?

1. // Demonstrates how to provide delegates to exectution dataflow blocks.
2. class DataflowExecutionBlocks
3. {
4. ????// 计算文件中包含零字节的总数
5. ????static
   int CountBytes(string path)
6. ????{
7. ????????byte[] buffer = new
   byte[1024];
8. ????????int totalZeroBytesRead = 0;
9. ????????using (var fileStream = File.OpenRead(path))
10. ????????{
11. ????????????int bytesRead = 0;
12. ????????????do
13. ????????????{
14. ????????????????bytesRead = fileStream.Read(buffer, 0, buffer.Length);
15. ????????????????totalZeroBytesRead += buffer.Count(b => b == 0);
16. ????????????} while (bytesRead > 0);
17. ????????}
18. ?
19. ????????return totalZeroBytesRead;
20. ????}
21. ?
22. ????static
    void Run(string[] args)
23. ????{
24. ????????// 创建一个临时目录
25. ????????string tempFile = Path.GetTempFileName();
26. ?
27. ????????// 随机写入数据
28. ????????using (var fileStream = File.OpenWrite(tempFile))
29. ????????{
30. ????????????Random rand = new Random();
31. ????????????byte[] buffer = new
    byte[1024];
32. ????????????for (int i = 0; i < 512; i++)
33. ????????????{
34. ????????????????rand.NextBytes(buffer);
35. ????????????????fileStream.Write(buffer, 0, buffer.Length);
36. ????????????}
37. ????????}
38. ?
39. ????????// 创建一个ActionBlock<int> 对象来打印 读取到的字节数
40. ????????var printResult = new ActionBlock<int>(zeroBytesRead =>
41. ????????{
42. ????????????Console.WriteLine("{0} contains {1} zero bytes.",
43. ????????????????Path.GetFileName(tempFile), zeroBytesRead);
44. ????????});
45. ?
46. ????????// 创基一个 TransformBlock<string, int>对象来调用CountBytes函数，并返回计算结果
47. ????????var countBytes = new TransformBlock<string, int>(
48. ????????????new Func<string, int>(CountBytes));
49. ?
50. ????????// 将两个块链接起来：TranformBlock<string,int>对象和ActionBlock对象。
51. ????????countBytes.LinkTo(printResult);
52. ?
53. ????????// 创建一个连续任务：当TransformBlock<string, int>完成时，通知打印结果已完成
54. ????????countBytes.Completion.ContinueWith(delegate { printResult.Complete(); });
55. ?
56. ????????// 输入临时目录
57. ????????countBytes.Post(tempFile);
58. ?
59. ????????// 标识结束
60. ????????countBytes.Complete();
61. ?
62. ????????// 等待打印完成
63. ????????printResult.Completion.Wait();
64. ?
65. ????????File.Delete(tempFile);
66. ????}
67. }

?

TransformBlock：一般用作传输和计算。类似函数式编程中的Map操作。

Func<TInput,TOutput>委托，通过Post，输入一个TInput参数。

Complete（），表明已完成

Completion任务，完成之后，当前任务结束。本质上来讲，TranformBlock运行在一个Task中。

?

ActionBlock：

Action<TInput> action委托，单一输入，执行诸如打印之类的操作。

同样也有Complete（）和Completion任务。

?

同时，以上两个均支持异步方法：

Transform

Action

?

1. var countBytesAsync = new TransformBlock<string, int>(async path =>
2. {
3. ????byte[] buffer = new
   byte[1024];
4. ????int totalZeroBytesRead = 0;
5. ????using (var fileStream = new FileStream(
6. ????????path, FileMode.Open, FileAccess.Read, FileShare.Read, 0x1000, true))
7. ????{
8. ????????int bytesRead = 0;
9. ????????do
10. ????????{
11. ????????????// Asynchronously read from the file stream.
12. ????????????bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length);
13. ????????????totalZeroBytesRead += buffer.Count(b => b == 0);
14. ????????} while (bytesRead > 0);
15. ????}
16. ?
17. ????return totalZeroBytesRead;
18. });