C#高级编程第11版

目录

• 导航
• 第十一章 Special Collections
  • 11.1 概述
  • 11.2 处理位
    • 11.2.1 BitArray 类
    • 11.2.2 BitVector32 结构
  • 11.3 可观察的集合
  • 11.4 不可变集合
    • 11.4.1 使用构建器来创建不可变集合
    • 11.4.2 支持不可变集合的类型和接口
    • 11.4.3 在不可变的数组中使用LINQ
  • 11.5 并发集合
    • 11.5.1 创建管道
    • 11.5.2 使用BlockingCollection
    • 11.5.3 使用ConcurrentDictionary
    • 11.5.4 完成管道
  • 11.6 小结
• 参考资料

导航

• C# 全版本特性一览
• 全书目录

第十一章 Special Collections

11.1 概述

在第10章，"集合"中，已经介绍了列表，队列，栈，字典和链表。本章将继续介绍一些特别的集合，譬如专门处理位数据的集合，可观察的集合，不变的集合以及能进行并发处理的集合等。

11.2 处理位

假如你需要对位数据进行处理，在第2章"核心C#"中我们介绍了C# 7开始可以使用"_"来使二进制数据看起来更加直观，而第6章"操作符与转换"中我们也介绍了各种位运算。你也可以使用BitArray类或者BitVector32结构体来对位数据进行处理。BitArray类位于System.Collections命名空间下，而BitVector32则位于System.Collections.Specialized。两者之间最重要的差别是——BitArray是变长的(resizable)，这点在你无法预计位数据的有几位的时候很有用，并且它可以存储更多位的位数据。BitVector32则是基于栈进行实现的，因此它的运行速度更快，只不过BitVector32最多只能保存32位的位数据。

11.2.1 BitArray 类

BitArray类是引用类型，它包含了一个int的数组，位数据每32位就会存在一个新的int整数中(for every 32 bits a new integer is used)。这个类的部分成员如下表所示：

成员	描述
Count Length	返回存储位的数组的长度。通过Length你还可以改变位数组的大小。
Item Get Set	你可以通过索引的方式来读写位数据。索引器是bool类型的，你也以 直接调用Get和Set方法来访问位数据。
SetAll	根据传入的参数直接设置所有位的数据。
Not	按位取反。
And Or Xor	通过这仨方法，你可以合并两个BitArray对象。

注意：第6章的时候，我们已经讲到位运算符可以用在各种数值类型上，如byte，short，int和long等等。BitArray类的功能也差不多，只不过比起C#基础类型来说，它可以处理更多数据位。

扩展方法GetBitsFormat遍历了整个BitArray并根据实际位设置输出相应的1和0到控制台上。为了可读性，我们在每4个位字符之间插入了"_"：

public static class BitArrayExtensions
{
	public static string GetBitsFormat(this BitArray bits)
	{
		var sb = new StringBuilder();
		for(int i = bits.Length - 1; i >= 0; i--)
		{
			sb.Append(bits[i] ? 1 : 0);
			if(i != 0 && i % 4 == 0)
			{
				sb.Append("_");
			}
		}
		return sb.ToString();
	}
}

在下面这个示例中，BitArray类创建了一个9位的bit，SetAll方法将所有位都设置位true。然后通过Set方法和索引两种模式将部分位置为false，最后我们调用上面的扩展方法，将它输出到控制台上：

var bits1 = new BitArray(9);
bits1.SetAll(true);
bits1.Set(1, false);
bits1[5] = false;
bits1[7] = false;
Console.Write("initialized: ");
Console.WriteLine(bits1.GetBitsFormat());

输出结果如下所示：

initialized: 1_0101_1101

Not方法将其按位取反，如果某位原先位true，取反之后就变成了false；反之原先为false，取反之后就变成了true：

Console.Write("not ");
Console.Write(bits1.GetBitsFormat());
bits1.Not();
Console.Write(" = ");
Console.WriteLine(bits1.GetBitsFormat());

输出结果如下所示：

not 1_0101_1101 = 0_1010_0010

接着我们用bits1作为参数传递给BitArray的构造函数来新建一个bits2，这样，它俩就拥有相同的值。接下来我们将bit2稍作修改，在调用Or方法之前，显示bits1和bits2当前的值。Or方法则改变了bits1的值：

var bits2 = new BitArray(bits1);
bits2[0] = true;
bits2[1] = false;
bits2[4] = true;
Console.Write($"{bits1.GetBitsFormat()} OR {bits2.GetBitsFormat()}");
Console.Write(" = ");
bits1.Or(bits2);
Console.WriteLine(bits1.GetBitsFormat());

输出结果如下所示：

0_1010_0010 OR 0_1011_0001 = 0_1011_0011

然后，我们再用And方法对bits2和bits1进行操作：

Console.Write($"{bits2.GetBitsFormat()} AND {bits1.GetBitsFormat()}");
Console.Write(" = ");
bits2.And(bits1);
Console.WriteLine(bits2.GetBitsFormat());

结果如下所示：

0_1011_0001 AND 0_1011_0011 = 0_1011_0001

最后，我们调用了Xor方法：

Console.Write($"{bits1.GetBitsFormat()} XOR {bits2.GetBitsFormat()}");
bits1.Xor(bits2);
Console.Write(" = ");
Console.WriteLine(bits1.GetBitsFormat());

如下所示：

0_1011_0011 XOR 0_1011_0001 = 0_0000_0010

11.2.2 BitVector32 结构

假如你事先就知道你要用多少位，你可以使用BitVector32结构体来代替BitArray类。BitVector32由于它是值类型，存储在栈上，所有的位都存在一个整数中，因此它更加地高效。只需要一个int整型你就可以放下32位的bit。如果你需要更多的位，你可以使用多个BitVector32或者直接使用BitArray类。BitArray类可以按需增长，但是BitVector32不行。

下面的表格列出了BitVector与BitArray不同的部分：

成员	描述
Data	返回一个int，用来代表BitVector32。
Item	你可以通过索引的方式获取BitVector32某位的值。 索引器有多个重载。你可以使用掩码或者内置的 BitVector32.Section作为索引。
CreateMask	静态方法。用来创建掩码。
CreateSection	静态方法。可以将32位的BitVector32分成若干个 Section。

这里我们补充一些代码，方便理解：

//Item...
public bool this[int bit]
{
	get => ((this.data & bit) == ((ulong) bit));
	set
	{
		if(value)
		{
			this.data |= (uint) bit;
		}
		else
		{
			this.data &= (uint) ~bit;
		}
	}
}
//CreateMask...
public static int CreateMask() => CreateMask(0);
public static int CreateMask(int previous)
{
	if(previous == 0)
	{
		return 1;
	}
	if(previous == -2147483648)
	{
		throw new InvalidOperationException(SR.GetString("BitVectorFull"));
	}
	return(previous << 1);
}

接下来我们回到书中的示例，如下所示：

var bits1 = new BitVector32();
int bit1 = BitVector32.CreateMask();
int bit2 = BitVector32.CreateMask(bit1);
int bit3 = BitVector32.CreateMask(bit2);
int bit4 = BitVector32.CreateMask(bit3);
int bit5 = BitVector32.CreateMask(bit4);
bits1[bit1] = true;
bits1[bit2] = false;
bits1[bit3] = true;
bits1[bit4] = true;
bits1[bit5] = true;
Console.WriteLine(bits1);

示例代码首先通过默认构造函数创建了一个BitVector32类型的变量bits1，默认情况下bits1的32位都初始化成false。接下来，依次创建掩码，以便访问到BitVector中内置的位。第一次调用CreateMask时，将创建可以访问首位(first bit)的掩码，bit1的值为1。每一次都将前一位的掩码作为参数传给CreateMask来得到下一位的掩码，bit2的值为2，而bit3的值则是4，bit4为8。

获得掩码之后，就可以作为索引，用来访问BitVector中内置的位，并设置相应的值。

BitVector32的ToString方法经过重写，可以显示每一位是否被设置，1代表设置，0表示没有，上面代码运行结果如下所示：

BitVector32{00000000000000000000000000011101}

除了通过CreateMask方法创建掩码，你也可以自己定义掩码；这样你可以一次性同时设置多个位。例如下面代码所示：

var bits1 = new BitVector32();
bits1[0xabcdef] = true;
Console.WriteLine(bits1);

十六进制的0xabcdef换算成二进制就是1010_1011_1100_1101_1110_1111。凡是1的位都被设置成了true，因此上面的代码运行后如下所示：

BitVector32{00000000101010111100110111101111} 

将32位bit分成若干个部分(Section)的特性特别地有用。举个例子，IPv4地址是由4个字节组成的(defined as a four-byte number)，它可以存储在一个整型里(inside an integer)。你可以通过定义4个Section来划分这个整型。在传播IP消息的时候，这些32位的数据就会被用上。32位数据还有一种应用情况：其中的16位被划成资源，8位用来存储查询者的请求码，3位用来校验(for a querier's robustness variable)，1位标志位，最后4位是保留位。当然你也可以根据你的实际需要来定义32位bit的存储方式。下面的例子就模拟了，接受一个值为0x79abcdef的变量received，并将其作为参数传递给BitVector32的构造函数：

int received = 0x79abcdef;
BitVector32 bits2 = new BitVector32(received);
Console.WriteLine(bits2);

bit2就被初始化成了received的值，结果如下所示：

BitVector32{01111001101010111100110111101111}

接下来我们创建6个节(Section)，如下所示：

// sections: FF EEE DDD CCCC BBBBBBBB AAAAAAAAAAAA
BitVector32.Section sectionA = BitVector32.CreateSection(0xfff);
BitVector32.Section sectionB = BitVector32.CreateSection(0xff, sectionA);
BitVector32.Section sectionC = BitVector32.CreateSection(0xf, sectionB);
BitVector32.Section sectionD = BitVector32.CreateSection(0x7, sectionC);
BitVector32.Section sectionE = BitVector32.CreateSection(0x7, sectionD);
BitVector32.Section sectionF = BitVector32.CreateSection(0x3, sectionE);

第一节需要12位，因此我们设置为0xfff(0b_1111_1111_1111)；第二节，8位；第三节，4位；第四和第五节3位，最后一节，2位。第一次调用CreateSection方法时仅仅根据参数0xfff分配了前面12位，而第二次开始调用的时候，将第一次分配的Section作为参数传递，即可在其后划分Section。CreateSection方法接收两个参数，Offset(移位)和Mask(掩码)并返回BitVector32.Section类型的值。

将BitVector32.Section作为索引类型将会返回一个int值，代表所在Section所有位对应的十进制int值，下面我们创建了一个扩展方法ToBinaryString，来返回这个int对应的二进制数据字符串：

public static string AddSeparators(this string number) 
	=> number.Length <= 4 ? number : 
		string.Join("_", Enumerable.Range(0, number.Length / 4)
			.Select(i => number.Substring(i * 4, 4)).ToArray());
public static string ToBinaryString(this int number) 
	=> Convert.ToString(number, toBase: 2).AddSeparators();

ToBinaryString接收一个int类型的数字作为参数，并且返回该数字仅包含0和1的二进制字符串。为了实现这一点，我们调用了Convert.ToString方法，并为其第二个参数toBase，传入了数字2，代表转换成二进制字符串。而通过AddSeparators方法，我们在其中调用了string.Join方法，来将数字的二进制字符串每4位之间插入一个"_"字符。这里我们还用到了两个LINQ方法，我们将在下一章中进行详细介绍。

将前面创建的6个Section作为索引传给bits2，主程序中的代码如下所示：

Console.WriteLine($"Section A:{ bits2[sectionA].ToBinaryString() }");
Console.WriteLine($"Section B:{ bits2[sectionB].ToBinaryString() }");
Console.WriteLine($"Section C:{ bits2[sectionC].ToBinaryString() }");
Console.WriteLine($"Section D:{ bits2[sectionD].ToBinaryString() }");
Console.WriteLine($"Section E:{ bits2[sectionE].ToBinaryString() }");
Console.WriteLine($"Section F:{ bits2[sectionF].ToBinaryString() }");

运行程序之后，输出如下所示：

Section A:1101_1110_1111
Section B:1011_1100
Section C:1010
Section D:1
Section E:111
Section F:1

11.3 可观察的集合

假如当集合元素增加或者移除时，你需要相关的信息，你可以使用ObservableCollection<T>类。这个类原本是在WPF里定义的，因此UI可以通知后台有些集合发生了变化。现在在通用Windows程序(UWP)中都可以使用。这个类的命名空间是System.Collections.ObjectModel。

ObservableCollection<T>类派生自Collection<T>，它可以用来创建个性化的集合(custom collections)，并且它在内部使用了List<T>。它重写了基类的SetItem和RemoveItem方法，以便触发CollectionChanged事件。调用者可以通过INotifyCollectionChanged接口来注册这个事件。

接下来的例子中使用了ObservableCollection<string>，并且Data_CollectionChanged方法被注册为CollectionChanged事件的处理方法。之后我们为它添加了两个元素，再插入第三个，移除第一个，代码如下所示：

var data = new ObservableCollection<string>();
data.CollectionChanged += Data_CollectionChanged;
data.Add("One");
data.Add("Two");
data.Insert(1, "Three");
data.Remove("One");

Data_CollectionChanged方法中会接收到NotifyCollectionChangedEventArgs事件参数，来获取集合中发生了哪些改变。参数的Action属性提供了元素是被添加还是移除的操作信息。当你移除某个元素的时候，你可以通过OldItems属性来访问它，而新增的元素则通过NewItems属性进行访问：

public static void Main()
{
	var data = new ObservableCollection < string > ();
	data.CollectionChanged += Data_CollectionChanged;
	data.Add("One");
	data.Add("Two");
	data.Insert(1, "Three");
	data.Remove("One");
}
public static void Data_CollectionChanged(object sender, NotifyCollectionChangedEventArgs e)
{
	Console.WriteLine($"action: {e.Action}");
	if(e.OldItems != null)
	{
		Console.WriteLine($"starting index for old item(s):{ e.OldStartingIndex}				");
		Console.WriteLine("old item(s):");
		foreach(var item in e.OldItems)
		{
			Console.WriteLine(item);
		}
	}
	if(e.NewItems != null)
	{
		Console.WriteLine($"starting index for new item(s):{ e.NewStartingIndex}");
		Console.WriteLine("new item(s): ");
		foreach(var item in e.NewItems)
		{
			Console.WriteLine(item);
		}
	}
	Console.WriteLine();
}

运行程序你将看到如下的输出：

action: Add
starting index for new item(s): 0
new item(s):
One
action: Add
starting index for new item(s): 1
new item(s):
Two
action: Add
starting index for new item(s): 1
new item(s):
Three
action: Remove
starting index for old item(s): 0
old item(s):
One

一开始集合中是空的，因此我们将One和Two俩个元素添加进集合的时候，显示的Action为Add，并且序号依次为0和1。而第三个元素Three执行的是插入，它被插入到序号为1的位置上，因此它显示的Action也是Add，序号为1。最后，我们将元素One移除，它是第一个，因此序号为0。

11.4 不可变集合

假如一个对象其状态可以发生变化，那么它就很难在多个同时运行的任务中共用。对这些集合来说同步很重要。假如一个对象的状态是恒定的，那么它就可以轻松地用于多线程。这种恒定不变的对象我们称之为不可变对象(immutable object)。恒定不变的集合我们称之为不可变集合(immutable collections)。

注意：我们将在第15章"异步编程"和第21章"任务和并行编程"，中详细介绍如何使用多任务和多线程进行异步编程。

让我们用一个简单的不可变数组ImmutableArray<string>来开始介绍，如下所示：

ImmutableArray<string> a1 = ImmutableArray.Create<string>();

你可以通过静态的Create方法来创建这个不可变数组。Create方法拥有多个重载版本，你可以给该方法传递任意s数量的参数。注意这里用到了两个不同的类型：非泛型版本的ImmutableArray类和它内置的Create方法，返回的是泛型版本的ImmutableArray结构体。在上面的代码里我们创建了一个空数组。

当然，一个空数组并没什么卵用。因此ImmutableArray<T>提供了Add方法来添加元素。然而与其它集合类不同的是，Add方法并未改变自身，而是返回添加元素后的新集合。因此，在调用Add方法之后，a1依然是个空集合，而a2则是拥有一个元素的不可变集合，如下所示：

ImmutableArray<string> a2 = a1.Add("Williams");

通过这一特性，使得你可以用一种很流利的方式(in a fluent way)来调用这个API。你可以挨个多次地调用Add方法。如下所示，a3现在引用的是一个拥有4个元素的不可变集合：

ImmutableArray<string> a3 = a2.Add("Ferrari").Add("Mercedes").Add("Red Bull Racing");

在使用这个不可变数组的每个阶段中，并不是每一步都会复制整个集合。实际上，不可变集合使用了共享状态(shared state)，只有在必要时，才会复制整个集合。

然而，如果事先就填充好整个集合，然后再将其转成不可变数组，将会更加高效。当遇到需要替换某些元素的操作时，你还可以将其恢复成可变集合。不可变类型中提供了构建器(builder)类来实现这一点。

为了演示这个过程，首先我们创建了一个Account类，该类自身也是不可变类型，所有的属性都是只读的：

public class Account
{
	public Account(string name, decimal amount)
	{
		Name = name;
		Amount = amount;
	}
	public string Name
	{
		get;
	}
	public decimal Amount
	{
		get;
	}
}

接下来我们创建了一个List<Account>集合，并为其填充一些示例数据：

var accounts = new List < Account > ()
{
	new Account("Scrooge McDuck", 667377678765m),
	new Account("Donald Duck", -200m),
	new Account("Ludwig von Drake", 20000m)
};

通过扩展方法ToImmutableList，从上面这个可变集合中，我们创建了一个不可变集合：

ImmutableList<Account> immutableAccounts = accounts.ToImmutableList();

不可变集合同样地可以进行遍历，它只是不允许被修改：

foreach (var account in immutableAccounts)
{
	Console.WriteLine($"{account.Name} {account.Amount}");
}

除了使用foreach语句来遍历不可变列表，你可以可以直接使用ImmutableList<T>自己定义的Foreach方法，该方法接收一个Action<T>类型的委托作为参数，因此你可以直接用lambda表达式为其传参：

immutableAccounts.ForEach(a => Console.WriteLine($"{a.Name} {a.Amount}"));

不可变集合还可以使用许多方法，如Contains，FindAll，FindLast，IndexOf等等。因为这些方法的使用与我们第10章介绍的普通集合大同小异，这里我们就不再展开介绍。

假如你需要修改不可变集合的内容，不可变集合同样提供了像Add，AddRange，Remove，RemoveAt，RemoveRange，Replace和Sort之类的方法。只不过，它们的实现与普通集合不同。调用这些方法后，不可变集合本身并不会发生任何改变，而是返回一个新的不可变集合。

11.4.1 使用构建器来创建不可变集合

通过上面我们提到的Add，Remove和Replace这些方法来从一个已有的不可变集合中，创建一个新的不可变集合是很简单，然而，当你需要多次频繁地对不可变集合进行增加或者删除的时候，这种方式的效率特别低。为了应对需要不可变集合多次修改的应用场景，你可以创建一个构建器(builder)。

让我们接着上面的示例代码，接下来我们将对其做大量的修改。为了实现这一点，你可以使用ToBuilder方法创建一个构建器，方法会返回一个可供修改的集合。在本例中，所有数额大于0的账户都被移除，在这个过程中，原始的不可变集合其实并没有发生变化。当构建器完成所有修改后，通过调用它的ToImmutable方法，一个新的不可变集合将被创建：

ImmutableList<Account>.Builder builder = immutableAccounts.ToBuilder();
for(int i = 0; i < builder.Count; i++)
{
	Account a = builder[i];
	if(a.Amount > 0)
	{
		builder.Remove(a);
	}
}
ImmutableList<Account> overdrawnAccounts = builder.ToImmutable();
overdrawnAccounts.ForEach(a => Console.WriteLine($"{a.Name} { a.Amount}"));

已经透支的账户将如下所示：

Donald Duck -200

除了移除元素的Remove方法以外，Builder类型还提供了如Add，AddRange，Insert，RemoveAt，RemoveAll，Reverse和Sort等方法。你只需要在修改操作完成后，调用ToImmutable来重新得到不可变集合就好了。

11.4.2 支持不可变集合的类型和接口

除了ImmutableArray和ImmutableList之外，NuGet包System.Collections.Immutable还提供了其它的不可变集合，部分类型如下表所示：

不可变类型	描述
ImmutableArray<T>	内部使用了Array类型并且不允许修改。 实现了IImmutableList<T>接口。
ImmutableList<T>	内部使用了二叉树来映射对象。 实现了IImmutableList<T>接口。
ImmutableQueue<T>	实现了IImmutableQueue<T>接口。允许 通过Enqueue，Dequeue和Peek访问元素。
ImmutableStack<T>	实现了IImmutableQueue<T>接口。允许 通过Push，Pop和Peek访问元素。
ImmutableDictionary<T>	实现了IImmutableDictionary<TKEy,TValue> 接口。保存的是无序的键值对。
ImmutableSortedDictionary<T>	同上，只不过键值对是有序的。
ImmutableHashSet<T>	实现了IImmutableSet<T>接口。保存的元素 是无序的。
ImmutableSortedSet<T>	同上，只不过存储的是有序的元素。

跟普通集合类一样，不可变集合类型也实现了一系列的接口，如IImmutableList<T>, IImmutableQueue<T>, 和IImmutableStack<T>等等。最大的区别在于不可变集合接口的方法实现的时候并不改变原有集合，而是返回修改后的新集合。

11.4.3 在不可变的数组中使用LINQ

ImmutableArrayExtensions为不可变数组定义了经过优化的LINQ方法，如Where，Aggregate，All，First，Last，Select和SelectMany等。如果你需要使用优化过的版本，你只需要引用System.Linq命名空间，直接使用ImmutableArray类型即可。

ImmutableArrayExtensions类为ImmutableArray扩展的方法大概如下所示(以Where方法为例)：

public static IEnumerable<T> Where<T>(this ImmutableArray<T> immutableArray, Func<T, bool> predicate);

ImmutableArray<T>其实也拥有普通版本的LINQ方法，这继承自IEnumerable<T>。但因为拥有更优配(better match)，因此在使用LINQ方法的时候会调用优化后的版本(optimized version)。

注意：第12章，"语言集合查询"中我们将会解释LINQ的细节。

11.5 并发集合

不可变集合可以很轻易地在多线程间使用，因为它们是不变的。而当你需要在多线程的时候使用可变集合，.NET提供了一些线程安全的集合类，它们定义在命名空间System.Collections.Concurrent下。线程安全的集合类保证了多线程在访问它们的时候不会导致冲突。

定义了接口IProducerConsumerCollection<T>用来保证访问集合类是线程安全的。该接口中最重要的两个方法是TryAdd和TryTake。TryAdd尝试为集合添加元素，但当集合处于锁状态的时候，这个操作可能会失败。该方法返回一个Boolean值来告知你添加操作是否成功。TryTake也是类似的逻辑，成功的时候返回success，并取到集合中的元素，失败的时候则会返回返回false值。下面的列表中描述了部分System.Collections.Concurrent下的类与它们的功能：

• ConcurrentQueue<T>：这个类通过无锁(lock-free)算法实现并且内部使用的是结合了32元素数组的链表(uses 32 item arrays that are combined in a linked list internally)。你可以使用Enqueue，TryQueue和TryPeek方法来访问队列元素。这些方法和你前面学习的Queue<T>其实很相似，只不过多了一个Try的前缀，以便在调用失败的时候可以返回布尔值告诉调用方。这个类也实现了IProducerConsumerCollection<T>接口，只不过它的TryAdd和TryTake方法调用的是Enqueue和TryDequeue而已。
• ConcurrentStack<T>：和ConcurrentQueue<T>非常的类似，只不过它拥有不同的访问元素的方法。这个类定义了Push，PushRange，TryPeek，TryPop和TryPopRange等方法。其内部元素也是用链表进行存储的。
• ConcurrentBag<T>：这个类并没有定义任何添加或者获取元素的顺序。它使用了一个概念，将多个线程映射成内部使用的数组(uses a concept that maps threads to arrays used internally)，来尝试减少对锁的使用。访问元素的方法有Add，TryPeek和TryTake等。
• ConcurrentDictionary<T>：这是一个线程安全的键值对集合。TryAdd，TryGetValue和TryUpdate方法用来在非阻塞的方式(in a nonblocking fashion)下访问内部成员。因为元素都是基于Key和Value的，因此这个类并不实现IProducerConsumerCollection<T>接口。
• BlockingCollection<T>：一种在轮到它处理添加或者获取元素之前，会自我阻塞并等待的线程。它提供一个接口，定义了Add和Take方法来添加移除元素。这些方法在任务可以执行之前会阻塞线程。Add方法有一个重载版本，借此(whereby)你可以传递一个CancellationToken类型的参数。这个令牌允许你取消一个阻塞中的调用(a blocking call)。假如你不想让线程无限等待，又或者你不想在外部手动取消阻塞，你可以使用TryAdd和TryTake方法。通过这俩方法，你可以指定一个超时时间，来决定在调用失败之前，你可以忍受线程被阻塞的最大时间。

ConcurrentXXX集合都是线程安全的，假如某个操作无法在线程当前线程上生效时，就会返回false。你需要在进行下一个操作之前先检查新增或移除元素是否成功。你不能相信每次集合都能完成任务(fulfill the task)。

BlockingCollection<T>对于任何实现了IProducerConsumerCollection<T>接口的类来说是一个装饰者(is a decorator)，并且它默认使用了ConcurrentDictionary<T>。在它的构造函数中，你也可以使用任何实现了IProducerConsumerCollection<T>接口的类型作为参数，譬如ConcurrentBag<T>和ConcurrentStack<T>。

11.5.1 创建管道

并发集合的一个经典应用就是管道。一个任务将某些内容写入集合里而另外一个任务可以同时从集合里读取数据。

接下来的示例中演示了如何在多任务中使用BlockingCollection<T>类来构建管道。管道第一部分的任务如下图所示：

第1阶段的任务是读取文件名并将它们添加到队列中，而在这个任务运行的时候，第2阶段也可以同时启动，它将从队列中读取文件名，并加载它们的内容。这些内容被写入到另外一个队列中。而第3阶段的任务是读取这些内容并进行处理，当然它也是同时启动的。在示例代码中，我们的处理是写入到一个字典中。

在整个场景中，第二部分只有在第3阶段完成所有处理并将所有的结果都写入到字典之后才会开始。后续的步骤如下图所示：

第4阶段读取上面完成的字典，转换它们的数据，并将其写入一个队列。第5阶段则是为这些元素添加色彩信息并将它们存入另外一个队列中，最后一个阶段则是将信息显示出来。4-6阶段也可以并发进行。

Info类代表管道保存的元素，代码如下所示：

public class Info
{
	public Info(string word, int count)
	{
		Word = word;
		Count = count;
	}
	public string Word
	{
		get;
	}
	public int Count
	{
		get;
	}
	public string Color
	{
		get;
		set;
	}
	public override string ToString() => $"{Count} times: {Word}";
}

主程序中主要是调用了StartPipelineAsync方法，如下所示：

public static void Main()
{
	StartPipelineAsync().Wait(); //这里用不用Wait结果没区别
	Console.ReadLine();
}

//具体每个阶段的方法在接下来的两小节中介绍
public static async Task StartPipelineAsync()
{
	var fileNames = new BlockingCollection < string > ();
	var lines = new BlockingCollection < string > ();
	var words = new ConcurrentDictionary < string, int > ();
	var items = new BlockingCollection < Info > ();
	var coloredItems = new BlockingCollection < Info > ();
	Task t1 = PipelineStages.ReadFilenamesAsync(@"../../..", fileNames);
	ColoredConsole.WriteLine("started stage 1");
	Task t2 = PipelineStages.LoadContentAsync(fileNames, lines);
	ColoredConsole.WriteLine("started stage 2");
	Task t3 = PipelineStages.ProcessContentAsync(lines, words);
	await Task.WhenAll(t1, t2, t3);
	ColoredConsole.WriteLine("stages 1, 2, 3 completed");
	Task t4 = PipelineStages.TransferContentAsync(words, items);
	Task t5 = PipelineStages.AddColorAsync(items, coloredItems);
	Task t6 = PipelineStages.ShowContentAsync(coloredItems);
	ColoredConsole.WriteLine("stages 4, 5, 6 started");
	await Task.WhenAll(t4, t5, t6);
	ColoredConsole.WriteLine("all stages finished");
}

在StartPipelineAsync方法中，我们实例化了各个并发集合，并将其作为参数传递给管道的不同阶段。前3个阶段ReadFilenamesAsync，LoadContentAsync和ProcessContentAsync是同时(simultaneously)运行的。而第4个阶段，则是需要等待前3个阶段完成后才可以开始。

注意：示例代码中使用了Task以及async和await关键字，这将在第15章进行介绍。你也可以在第21章中了解更多关于线程，任何和同步的讯息。文件的I/O处理则是在第22章，"文件和流"中介绍。

示例代码中使用了自定义的ColoredConsole类来向控制台输出带色彩的信息。这个类提供了一种简单的保证同步的方式(指使用lock关键字)，来避免输出了错误的颜色：

public static class ColoredConsole
{
	private static object syncOutput = new object();
	public static void WriteLine(string message)
	{
		lock(syncOutput)
		{
			Console.WriteLine(message);
		}
	}
	public static void WriteLine(string message, string color)
	{
		lock(syncOutput)
		{
			Console.ForegroundColor = (ConsoleColor) Enum.Parse(typeof(ConsoleColor), color);
			Console.WriteLine(message);
			Console.ResetColor();
		}
	}
}

11.5.2 使用BlockingCollection

让我们先来处理管道的第一部分。在第1阶段中，ReadFilenamesAsync方法接收一个BlockingCollection<T>类型的参数，它可以将读取到的文件名输出到其上。具体代码如下所示：

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

通过遍历指定路径下所有的目录和子目录，查找到后缀名为cs的文件，并通过Add方法添加到BlockingCollection中。当所有cs文件都添加完成后，调用CompleteAdding方法来通知调用方不需要再等待任何元素的添加。

注意：假如你除了往BlockingCollection中写入数据的时候，同时还有一个任务负责从它里面读取数据，那么CompleteAdding的调用就至关重要了。否则，负责读取数据的任务将会因为可能还会有新数据添加而无限等待下去。

第2阶段是由LoadContentAsync方法来完成的，它将读取文件内容并存储到另外一个集合中。代码如下所示：

//注意6个阶段中，仅有第2阶段使用了await关键字
//并且没有使用Task.Factory.StartNew()的方式来写
//具体到后面章节再进行解释
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();
}

这个方法从输入的BlockingCollection中获取到文件名，打开相应的文件，读取文件中所有行，并添加到输出的BlockingCollection中。foreach语句中通过输入的BlockingCollection的GetConsumingEnumerable方法来遍历其所拥有的元素。可能你会直接使用变量input而非调用它的GetConsumingEnumerable方法，但直接使用的话，仅能遍历集合当前的状态，之后其它任务再往集合中添加的元素，则无法被访问到。

注意：假如你读取的BlockingCollection同时会被其它任务填充，那么你需要使用GetConsumingEnumerable方法才能获取到实时的BlockingCollection元素。

11.5.3 使用ConcurrentDictionary

第3阶段我们通过ProcessContentAsync方法实现。代码如下所示：

public static Task ProcessContentAsync(BlockingCollection < string > input, ConcurrentDictionary < string, int > output)
{
	return Task.Factory.StartNew(() =>
	{
		foreach(var line in input.GetConsumingEnumerable())
		{
			string[] words = line.Split(' ', ';', '	', '{', '}', '(', ')', ':', ',', '"');
			foreach(var word in words.Where(w => !string.IsNullOrEmpty(w)))
			{
				output.AddOrUpdate(key: word, addValue: 1, updateValueFactory: (s, i) => ++i);
				ColoredConsole.WriteLine($"stage 3: added {word}");
			}
		}
	}, TaskCreationOptions.LongRunning);
}

方法从输入的BlockingCollection那获取到文件行的内容，然后按照指定的分隔符将其分割成单词，并存入输出的字典中。方法AddOrUpdate是ConcurrentDictionary类型定义的方法(普通Dictionary是没有的)。假如指定的Key值还不存在于字典中，第二个参数定义了将为它存储的值。假如某个Key已经存在于字典中，第三个参数updateValueFactory则定义了应该如何修改已存在的字典值。在本例中，我们让它自增1。

运行程序，你可能会看到这样的输出，第2阶段和第3阶段随机交替出现：

...
stage 2: added         }
stage 2: added     }
stage 2: added }
stage 3: added using
stage 3: added System
stage 3: added using
stage 2: added using System;
stage 2: added using System.Collections;
stage 3: added System.Runtime.Serialization
stage 3: added namespace
stage 3: added Wrox.ProCSharp.Delegates
stage 3: added [Serializable]
stage 2: added using System.Collections.Concurrent;
stage 3: added internal
...

11.5.4 完成管道

当前面3个阶段完成之后，接下来的3个阶段也可以再次并发运行。TransferContentAsync从字典中获取数据，并将其转换成Info类型，然后再将它输出到BlockingCollection中：

public static Task TransferContentAsync(ConcurrentDictionary < string, int > input, BlockingCollection < Info > output)
{
	return Task.Factory.StartNew(() =>
	{
		foreach(var word in input.Keys)
		{
			if(input.TryGetValue(word, out int value))
			{
				var info = new Info(word, value); 				
				output.Add(info);
				ColoredConsole.WriteLine($"stage 4: added {info}");
			}
		}
		output.CompleteAdding();
	}, TaskCreationOptions.LongRunning);
}

AddColorAsync则是负责根据Info类型实例的Count属性来设置它的Color属性：

public static Task AddColorAsync(BlockingCollection < Info > input, BlockingCollection < Info > output)
{
	return Task.Factory.StartNew(() =>
	{
		foreach(var item in input.GetConsumingEnumerable())
		{
			if(item.Count > 40)
			{
				item.Color = "Red";
			}
			else if(item.Count > 20)
			{
				item.Color = "Yellow";
			}
			else
			{
				item.Color = "Green";
			}
			output.Add(item);
			ColoredConsole.WriteLine($"stage 5: added color{ item.Color } to { item	}");
		}
		output.CompleteAdding();
	}, TaskCreationOptions.LongRunning);
}

最后一个阶段则是将结果按照指定颜色输出到控制台上：

public static Task ShowContentAsync(BlockingCollection < Info > input)
{
	return Task.Factory.StartNew(() =>
	{
		foreach(var item in input.GetConsumingEnumerable())
		{
			ColoredConsole.WriteLine($"stage 6: {item}", item.Color);
		}
	}, TaskCreationOptions.LongRunning);
}

这样，当你运行整个程序的时候，你会看见控制台上是彩色的输出，如下所示：

11.6 小结

本章着眼于如何使用(work with)特殊集合。本章为你介绍了BitArray类和BitVector32结构体，它们优化了处理bit数据的方式。

不仅仅只有bit数据可以存储在可观察集合ObservableCollection<T>中。当列表中的元素发生修改时，可观察集合将会触发特定的事件。第33章到37章将使用这个类来创建Windows和Xamarin应用。

本章也解释了不可变集合是如何来保障集合是不变的，因此它们可以很轻松地用于多线程环境中。

本章最后一部分则着眼于如何使用并发集合，当一个线程在往集合中写入数据时，其它的线程还可以同时从同一个集合中获取(retrieve)它的元素。

第12章我们将介绍语言集成查询LINQ的细节。

参考资料

• C# ConcurrentBag的实现原理