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. });