最近写了个小程序用到了C#4.0中的线程安全集合。想起很久以前用C#2.0开发的时候写后台windows服务，为了利用多线程实现生产者和消费者模型，经常要封装一些线程安全的容器，比如泛型队列和字典等等。下面就结合部分MS的源码和自己的开发经验浅显地分析一下如何实现线程安全容器以及实现线程安全容器容易产生的问题。
在C#早期版本中已经实现了线程安全的ArrayList，可以通过下面的方式构造线程安全的数组列表：
var array = ArrayList.Synchronized(new ArrayList());我们从Synchronized方法入手，分析它的源代码看是如何实现线程安全的：
Synchronized /// <summary>Returns an <see cref="T:System.Collections.ArrayList" /> wrapper that is synchronized (thread safe).</summary>
/// <returns>An <see cref="T:System.Collections.ArrayList" /> wrapper that is synchronized (thread safe).</returns>
/// <param name="list">The <see cref="T:System.Collections.ArrayList" /> to synchronize. </param>
/// <exception cref="T:System.ArgumentNullException">
/// <paramref name="list" /> is null. </exception>
/// <filterpriority>2</filterpriority>
[HostProtection(SecurityAction.LinkDemand, Synchronization = true)]
public static ArrayList Synchronized(ArrayList list)
{
if (list == null)
{
throw new ArgumentNullException("list");
}
return new ArrayList.SyncArrayList(list);
}
继续跟进去，发现SyncArrayList是一个继承自ArrayList的私有类，内部线程安全方法的实现经过分析，很多都是像下面这样lock（注意是lock_root对象而不是数组列表实例对象）一下完事：
lock (this._root)
有心的你可以查看SyncArrayList的源码：
SyncArrayList [Serializable]
private class SyncArrayList : ArrayList
{
private ArrayList _list;
private object _root;
public override int Capacity
{
get
{
int capacity;
lock (this._root)
{
capacity = this._list.Capacity;
}
return capacity;
}
set
{
lock (this._root)
{
this._list.Capacity = value;
}
}
}
public override int Count
{
get
{
int count;
lock (this._root)
{
count = this._list.Count;
}
return count;
}
}
public override bool IsReadOnly
{
get
{
return this._list.IsReadOnly;
}
}
public override bool IsFixedSize
{
get
{
return this._list.IsFixedSize;
}
}
public override bool IsSynchronized
{
get
{
return true;
}
}
public override object this[int index]
{
get
{
object result;
lock (this._root)
{
result = this._list[index];
}
return result;
}
set
{
lock (this._root)
{
this._list[index] = value;
}
}
}
public override object SyncRoot
{
get
{
return this._root;
}
}
internal SyncArrayList(ArrayList list)
: base(false)
{
this._list = list;
this._root = list.SyncRoot;
}
public override int Add(object value)
{
int result;
lock (this._root)
{
result = this._list.Add(value);
}
return result;
}
public override void AddRange(ICollection c)
{
lock (this._root)
{
this._list.AddRange(c);
}
}
public override int BinarySearch(object value)
{
int result;
lock (this._root)
{
result = this._list.BinarySearch(value);
}
return result;
}
public override int BinarySearch(object value, IComparer comparer)
{
int result;
lock (this._root)
{
result = this._list.BinarySearch(value, comparer);
}
return result;
}
public override int BinarySearch(int index, int count, object value, IComparer comparer)
{
int result;
lock (this._root)
{
result = this._list.BinarySearch(index, count, value, comparer);
}
return result;
}
public override void Clear()
{
lock (this._root)
{
this._list.Clear();
}
}
public override object Clone()
{
object result;
lock (this._root)
{
result = new ArrayList.SyncArrayList((ArrayList)this._list.Clone());
}
return result;
}
public override bool Contains(object item)
{
bool result;
lock (this._root)
{
result = this._list.Contains(item);
}
return result;
}
public override void CopyTo(Array array)
{
lock (this._root)
{
this._list.CopyTo(array);
}
}
public override void CopyTo(Array array, int index)
{
lock (this._root)
{
this._list.CopyTo(array, index);
}
}
public override void CopyTo(int index, Array array, int arrayIndex, int count)
{
lock (this._root)
{
this._list.CopyTo(index, array, arrayIndex, count);
}
}
public override IEnumerator GetEnumerator()
{
IEnumerator enumerator;
lock (this._root)
{
enumerator = this._list.GetEnumerator();
}
return enumerator;
}
public override IEnumerator GetEnumerator(int index, int count)
{
IEnumerator enumerator;
lock (this._root)
{
enumerator = this._list.GetEnumerator(index, count);
}
return enumerator;
}
public override int IndexOf(object value)
{
int result;
lock (this._root)
{
result = this._list.IndexOf(value);
}
return result;
}
public override int IndexOf(object value, int startIndex)
{
int result;
lock (this._root)
{
result = this._list.IndexOf(value, startIndex);
}
return result;
}
public override int IndexOf(object value, int startIndex, int count)
{
int result;
lock (this._root)
{
result = this._list.IndexOf(value, startIndex, count);
}
return result;
}
public override void Insert(int index, object value)
{
lock (this._root)
{
this._list.Insert(index, value);
}
}
public override void InsertRange(int index, ICollection c)
{
lock (this._root)
{
this._list.InsertRange(index, c);
}
}
public override int LastIndexOf(object value)
{
int result;
lock (this._root)
{
result = this._list.LastIndexOf(value);
}
return result;
}
public override int LastIndexOf(object value, int startIndex)
{
int result;
lock (this._root)
{
result = this._list.LastIndexOf(value, startIndex);
}
return result;
}
public override int LastIndexOf(object value, int startIndex, int count)
{
int result;
lock (this._root)
{
result = this._list.LastIndexOf(value, startIndex, count);
}
return result;
}
public override void Remove(object value)
{
lock (this._root)
{
this._list.Remove(value);
}
}
public override void RemoveAt(int index)
{
lock (this._root)
{
this._list.RemoveAt(index);
}
}
public override void RemoveRange(int index, int count)
{
lock (this._root)
{
this._list.RemoveRange(index, count);
}
}
public override void Reverse(int index, int count)
{
lock (this._root)
{
this._list.Reverse(index, count);
}
}
public override void SetRange(int index, ICollection c)
{
lock (this._root)
{
this._list.SetRange(index, c);
}
}
public override ArrayList GetRange(int index, int count)
{
ArrayList range;
lock (this._root)
{
range = this._list.GetRange(index, count);
}
return range;
}
public override void Sort()
{
lock (this._root)
{
this._list.Sort();
}
}
public override void Sort(IComparer comparer)
{
lock (this._root)
{
this._list.Sort(comparer);
}
}
public override void Sort(int index, int count, IComparer comparer)
{
lock (this._root)
{
this._list.Sort(index, count, comparer);
}
}
public override object[] ToArray()
{
object[] result;
lock (this._root)
{
result = this._list.ToArray();
}
return result;
}
public override Array ToArray(Type type)
{
Array result;
lock (this._root)
{
result = this._list.ToArray(type);
}
return result;
}
public override void TrimToSize()
{
lock (this._root)
{
this._list.TrimToSize();
}
}
}
同样，在C#早期版本中实现了线程安全的Hashtable，它也是早期开发中经常用到的缓存容器，可以通过下面的方式构造线程安全的哈希表：
var ht = Hashtable.Synchronized(new Hashtable());同样地，我们从Synchronized方法入手，分析它的源代码看是如何实现线程安全的：
lock (this._table.SyncRoot)
贴一下SyncHashtable的源码：
从上面的实现分析来说，封装一个线程安全的容器看起来并不是什么难事，除了对线程安全容器的异常处理心有余悸，其他的似乎按步就班就可以了，不是吗？也许还有更高明的实现吧？
在4.0中，多了一个System.Collections.Concurrent命名空间，怀着忐忑的心情查看C#4.0其中的一个线程安全集合ConcurrentQueue的源码，发现它继承自IProducerConsumerCollection<T>, IEnumerable<T>, ICollection, IEnumerable接口，内部实现线程安全的时候，通过SpinWait和通过互锁构造（Interlocked）及SpinWait封装的Segment，间接实现了线程安全。Segment的实现比较复杂，和线程安全密切相关的方法就是TryXXX那几个方法，源码如下：
和ArrayList以及Hashtable线程安全的“曲折”实现有点不同，ConcurrentQueue<T>一开始就是朝着线程安全方向实现去的。它没有使用lock，因为大家知道使用lock性能略差，对于读和写操作，应该分开，不能一概而论。ConcurrentQueue<T>具体实现在性能和异常处理上应该已经考虑的更全面周到一点。
在我看来，ConcurrentQueue<T>线程安全的具体实现有多吸引人在其次，IProducerConsumerCollection<T>接口的抽象和提取非常值得称道，查看源码发现ConcurrentStack<T>和ConcurrentBag<T>也继承自该接口。<<CLR via C#>>一书中在谈到接口和抽象类的时候特别举了集合和流(Stream)的例子，微软为什么如此设计，想起来果然很有深意。
对于线程安全的泛型字典ConcurrentDictionary<TKey, TValue>，我们也可以查看它的源码看它的具体实现方式。看源码有1200多行，实现稍微复杂一些。我们仅从最简单的TryAdd方法分析：
四、如法炮制
如果让我来构造实现线程安全容器，最简单直接快速高效的方式就是参考ArrayList和 Hashtable，我们完全可以模仿它们的处理方式，通过继承一个容器，然后内部通过lock一个SyncRoot对象，中规中矩地实现framework中其他容器的线程安全。比如要实现线程安全的泛型队列Queue<T>，贴一下大致的伪代码：
private Queue<T> queue = null;
private object syncRoot = null;
internal object SyncRoot
{
get
{
return syncRoot;
}
}
#endregion
#region constructors
public SyncQueue()
{
syncRoot = new object();
queue = new Queue<T>();
}
public SyncQueue(IEnumerable<T> collection)
{
syncRoot = new object();
queue = new Queue<T>(collection);
}
public SyncQueue(int capacity)
{
syncRoot = new object();
queue = new Queue<T>(capacity);
}
#endregion
#region methods
public new void Enqueue(T item)
{
lock (SyncRoot)
{
this.Enqueue(item);
}
}
public new T Dequeue()
{
T result = default(T);
lock (SyncRoot)
{
result = this.queue.Dequeue();
}
return result;
}
public new void Clear()
{
lock (SyncRoot)
{
this.queue.Clear();
}
}
public new bool Contains(T item)
{
var exists = false;
lock (SyncRoot)
{
exists = this.queue.Contains(item);
}
return exists;
}
#endregion
}
你可能觉得上面这样不动脑的方式似乎很傻很天真，但这绝对是一种正常人都能想到的思路，谁让MS的数组列表和哈希表就是这么实现的呢？
当然，我们还能想到的一种常见实现方式就是通过组合而不是类继承，实现的伪代码类似下面这样：
private Queue<T> queue = null;
private object syncRoot = null;
internal object SyncRoot
{
get
{
return syncRoot;
}
}
#endregion
#region constructors
public SyncQueue()
{
syncRoot = new object();
queue = new Queue<T>();
}
public SyncQueue(IEnumerable<T> collection)
{
syncRoot = new object();
queue = new Queue<T>(collection);
}
public SyncQueue(int capacity)
{
syncRoot = new object();
queue = new Queue<T>(capacity);
}
#endregion
#region methods
public void Enqueue(T item)
{
lock (SyncRoot)
{
this.Enqueue(item);
}
}
public T Dequeue()
{
T result = default(T);
lock (SyncRoot)
{
result = this.queue.Dequeue();
}
return result;
}
public void Clear()
{
lock (SyncRoot)
{
this.queue.Clear();
}
}
public bool Contains(T item)
{
var exists = false;
lock (SyncRoot)
{
exists = this.queue.Contains(item);
}
return exists;
}
#endregion
}
到这里，我们至少可以分析得出，实现一般的线程安全容器的思路至少有两种：类继承（内部实现偏向使用组合）和（或）组合，线程安全的地方只要通过framework的同步构造如lock、Interlocked等实现即可。
思考：如果让您实现线程安全容器，您优先会怎么实现呢？
CacheUtil缓存实现的伪代码如下:
public static bool TryAdd(object key, object value)
{
ht[key] = value; //set方法是线程安全的
return true;
}
public static bool TryGet(object key, out object result)
{
result = null;
lock (ht.SyncRoot)
{
if (ht.ContainsKey(key))
{
result = ht[key];
}
}
return true;
}
}
从代码中可以看出来，哈希表中的Value存放的是IList类型，那么值所保存的应该是一个引用（也就是指针）。
（1）、当线程1通过索引器得到这个IList时，这个TryGet读取操作是线程安全的。接着线程1进行的操作是列表遍历。在foreach进行遍历不为空的List的时候，遍历的其实是存放在IList指针指向的引用。
（2）、在foreach遍历集合的时候，这时候线程2如果正好对哈希表的key所对应的Value进行修改，IList的指针所指向的引用改变了，所以线程1的遍历操作就会抛出异常。
这是一个简单而又经典的陷阱，在哈希表的MSDN线程安全块有一段说明：
Enumerating through a collection is intrinsically not a thread safe procedure. Even when a collection is synchronized, other threads can still modify the collection, which causes the enumerator to throw an exception. To guarantee thread safety during enumeration, you can either lock the collection during the entire enumeration or catch the exceptions resulting from changes made by other threads.
列表通过索引取值，一个简单的示例代码如下：
按照类似于1中的分析，GetFirstOrDefault应该可以分为下面两步：
（1）线程1取数据，判断list.Count的时候发现列表内有1个元素，这一步线程安全，没有任何问题，然后准备返回索引为0的元素；
（2）线程2在线程1将要取索引为0的元素之前移除了列表中的唯一元素或者直接将list指向null，这样线程1通过索引取元素就抛出异常了。
从上面的两个示例，我们得知通常所看到的线程安全实际上并不一定安全。不安全的主要原因就是容器内的数据很容易被其他线程改变，或者可以简要概括为：一段时间差引发的血案。实际上，我们平时所做的业务系统，归根结底很多bug或者隐藏的缺陷都是由不起眼的一小段时间差引起的。
保证容器内的数据和操作都安全，一种简单而有效的方法就是将你所要进行的操作进行“事务”处理。比如示例1中哈希表的Value的遍历操作，通常情况下，我们分作两步：
（1）、（安全地）读取数据
（2）、（不安全地）遍历；
为了达到遍历操作不抛出异常，我们可以把两步合并为一步，抽象出一个线程安全的新方法TryGetAndEnumerate，这样可以保证线程安全地取数据和遍历，具体实现无非是lock一下SyncRoot类似的这种思路。但是这种线程安全的遍历可能代价很高，而且极其不通用。
线程安全集合容易产生的问题和解决方法，请参考JaredPar MSFT的Why are thread safe collections so hard?，这篇文章对设计一个线程安全的容器的指导原则是：
1、Don't add an decision procedures(procedures like Count as decision procedures). They lead users down the path to bad code.
2、Methods which query the object can always fail and the API should reflect this.
实际上大家都知道利用事务处理思想多用TryXXX方法一般是没错的。
新闻热点
疑难解答
