C# 使用 BlockingCollection

现在介绍管道的第一阶段。ReadFilenamesAsync 接收 BlockingCollection<T>为参数，在其中写入输出。该方法的实现使用枚举器来迭代指定目录及其子目录中的C# 文件。这些文件的文件名用 Add 方法添加到 BlockingCollection<T>中。完成添加文件名的操作后，调用 CompleteAdding 方法，以通知所有读取器不应再等待集合中的任何额外项：.

public static class PipelineStages{  public static Task ReadFilenamesAsync(string path,     BlockingCollection<string> output)  {    return Task.Factory.StartNew(() =>    {      foreach (string filename in Directory.EnumerateFiles(path,"*.cs",        SearchOption.AllDirectories))      {        output.Add(filename);        ColoredConsole.WriteLine($"stage 1: added {filename}");      }      output.CompleteAdding();    }，TaskCreationOptions.LongRunning);  }  //...

注意

如果在写入器添加项的同时，读取器从 BlockingCollection<T>中读取，那么调用CompleteAdding 方法是很重要的。否则，读取器会在 foreach 循环中等待更多的项被添加。

下一个阶段是读取文件并将其内容添加到另一个集合中，这由 LoadContentAsync 方法完成。该方法使用了输入集合传递的文件名，打开文件，然后把文件中的所有行添加到输出集合中。在 foreach 循环中，用输入阻塞集合调用GetConsumingEnumerable，以迭代各项。直接使用 input 变量而不调用GetConsumingEnumerable 是可以的，但是这只会迭代当前状态的集合，而不会迭代以后添加的项。

public static async Task LoadContentAsync(BlockingCollection<string> input,   BlockingCollection<string> output){  foreach (var filename in input.GetConsumingEnumerable())   {    using (FileStream stream = File.OpenRead(filename))    {      var reader = new StreamReader(stream);       string line = null;      while ((line = await reader.ReadLineAsync()) != null)      {        output.Add(line);        ColoredConsole.WriteLine($"stage 2: added {line}");      }    }  }  output.CompleteAdding();}

注意

如果在填充集合的同时，使用读取器读取集合，则需要使用GetConsumingEnumerable 方法获取阻塞集合的枚举器，而不是直接选代集合。

董川民