深入线程安全容器的实现方法
最近写了个小程序用到了c#4.0中的线程安全集合。想起很久以前用c#2.0开发的时候写后台windows服务，为了利用多线程实现生产者和消费者模型，经常要封装一些线程安全的容器，比如泛型队列和字典等等。下面就结合部分ms的源码和自己的开发经验浅显地分析一下如何实现线程安全容器以及实现线程安全容器容易产生的问题。
一、arraylist
在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();
}
}
}
arraylist线程安全实现过程小结：定义arraylist的私有实现syncarraylist，子类内部通过lock同步构造实现线程安全，在arraylist中通过synchronized对外间接调用子类。
二、hashtable
同样，在c#早期版本中实现了线程安全的hashtable，它也是早期开发中经常用到的缓存容器，可以通过下面的方式构造线程安全的哈希表：
var ht = hashtable.synchronized(new hashtable());
同样地，我们从synchronized方法入手，分析它的源代码看是如何实现线程安全的：
synchronized /// <summary>returns a synchronized (thread-safe) wrapper for the <see cref="t:system.collections.hashtable" />.</summary>
/// <returns>a synchronized (thread-safe) wrapper for the <see cref="t:system.collections.hashtable" />.</returns>
/// <param name="table">the <see cref="t:system.collections.hashtable" /> to synchronize. </param>
/// <exception cref="t:system.argumentnullexception">
/// <paramref name="table" /> is null. </exception>
/// <filterpriority>1</filterpriority>
[hostprotection(securityaction.linkdemand, synchronization = true)]
public static hashtable synchronized(hashtable table)
{
if (table == null)
{
throw new argumentnullexception("table");
}
return new hashtable.synchashtable(table);
}
继续跟进去，发现synchashtable是一个继承自hashtable和ienumerable接口的私有类，内部线程安全方法的实现经过分析，很多都是像下面这样lock（注意是lock哈希表的syncroot object实例对象而不是哈希表实例）一下完事：
lock (this._table.syncroot)
贴一下synchashtable的源码：
[serializable]
private class synchashtable : hashtable, ienumerable
{
protected hashtable _table;
public override int count
{
get
{
return this._table.count;
}
}
public override bool isreadonly
{
get
{
return this._table.isreadonly;
}
}
public override bool isfixedsize
{
get
{
return this._table.isfixedsize;
}
}
public override bool issynchronized
{
get
{
return true;
}
}
public override object this[object key]
{
[targetedpatchingoptout("performance critical to inline across ngen image boundaries")]
get
{
return this._table[key];
}
set
{
lock (this._table.syncroot)
{
this._table[key] = value;
}
}
}
public override object syncroot
{
get
{
return this._table.syncroot;
}
}
public override icollection keys
{
get
{
icollection keys;
lock (this._table.syncroot)
{
keys = this._table.keys;
}
return keys;
}
}
public override icollection values
{
get
{
icollection values;
lock (this._table.syncroot)
{
values = this._table.values;
}
return values;
}
}
internal synchashtable(hashtable table)
: base(false)
{
this._table = table;
}
internal synchashtable(serializationinfo info, streamingcontext context)
: base(info, context)
{
this._table = (hashtable)info.getvalue("parenttable", typeof(hashtable));
if (this._table == null)
{
throw new serializationexception(environment.getresourcestring("serialization_insufficientstate"));
}
}
[securitycritical]
public override void getobjectdata(serializationinfo info, streamingcontext context)
{
if (info == null)
{
throw new argumentnullexception("info");
}
lock (this._table.syncroot)
{
info.addvalue("parenttable", this._table, typeof(hashtable));
}
}
public override void add(object key, object value)
{
lock (this._table.syncroot)
{
this._table.add(key, value);
}
}
public override void clear()
{
lock (this._table.syncroot)
{
this._table.clear();
}
}
[targetedpatchingoptout("performance critical to inline across ngen image boundaries")]
public override bool contains(object key)
{
return this._table.contains(key);
}
[targetedpatchingoptout("performance critical to inline across ngen image boundaries")]
public override bool containskey(object key)
{
return this._table.containskey(key);
}
public override bool containsvalue(object key)
{
bool result;
lock (this._table.syncroot)
{
result = this._table.containsvalue(key);
}
return result;
}
public override void copyto(array array, int arrayindex)
{
lock (this._table.syncroot)
{
this._table.copyto(array, arrayindex);
}
}
public override object clone()
{
object result;
lock (this._table.syncroot)
{
result = hashtable.synchronized((hashtable)this._table.clone());
}
return result;
}
ienumerator ienumerable.getenumerator()
{
return this._table.getenumerator();
}
public override idictionaryenumerator getenumerator()
{
return this._table.getenumerator();
}
public override void remove(object key)
{
lock (this._table.syncroot)
{
this._table.remove(key);
}
}
public override void ondeserialization(object sender)
{
}
internal override keyvaluepairs[] tokeyvaluepairsarray()
{
return this._table.tokeyvaluepairsarray();
}
}
hashtable线程安全实现过程小结：定义hashtable的私有实现synchashtable，子类内部通过lock同步构造实现线程安全，在hashtable中通过synchronized对外间接调用子类。
三、4.0中的线程安全容器
1、concurrentqueue
从上面的实现分析来说，封装一个线程安全的容器看起来并不是什么难事，除了对线程安全容器的异常处理心有余悸，其他的似乎按步就班就可以了，不是吗？也许还有更高明的实现吧？
在4.0中，多了一个system.collections.concurrent命名空间，怀着忐忑的心情查看c#4.0其中的一个线程安全集合concurrentqueue的源码，发现它继承自iproducerconsumercollection<t>, ienumerable<t>, icollection, ienumerable接口，内部实现线程安全的时候，通过spinwait和通过互锁构造（interlocked）及spinwait封装的segment，间接实现了线程安全。segment的实现比较复杂，和线程安全密切相关的方法就是tryxxx那几个方法，源码如下：
private class segment
{
internal t[] m_array;
private int[] m_state;
private concurrentqueue<t>.segment m_next;
internal readonly long m_index;
private int m_low;
private int m_high;
internal concurrentqueue<t>.segment next
{
get
{
return this.m_next;
}
}
internal bool isempty
{
get
{
return this.low > this.high;
}
}
internal int low
{
get
{
return math.min(this.m_low, 32);
}
}
internal int high
{
get
{
return math.min(this.m_high, 31);
}
}
internal segment(long index)
{
this.m_array = new t[32];
this.m_state = new int[32];
this.m_high = -1;
this.m_index = index;
}
internal void unsafeadd(t value)
{
this.m_high++;
this.m_array[this.m_high] = value;
this.m_state[this.m_high] = 1;
}
internal concurrentqueue<t>.segment unsafegrow()
{
concurrentqueue<t>.segment segment = new concurrentqueue<t>.segment(this.m_index + 1l);
this.m_next = segment;
return segment;
}
internal void grow(ref concurrentqueue<t>.segment tail)
{
concurrentqueue<t>.segment next = new concurrentqueue<t>.segment(this.m_index + 1l);
this.m_next = next;
tail = this.m_next;
}
internal bool tryappend(t value, ref concurrentqueue<t>.segment tail)
{
if (this.m_high >= 31)
{
return false;
}
int num = 32;
try
{
}
finally
{
num = interlocked.increment(ref this.m_high);
if (num <= 31)
{
this.m_array[num] = value;
this.m_state[num] = 1;
}
if (num == 31)
{
this.grow(ref tail);
}
}
return num <= 31;
}
internal bool tryremove(out t result, ref concurrentqueue<t>.segment head)
{
spinwait spinwait = default(spinwait);
int i = this.low;
int high = this.high;
while (i <= high)
{
if (interlocked.compareexchange(ref this.m_low, i + 1, i) == i)
{
spinwait spinwait2 = default(spinwait);
while (this.m_state[i] == 0)
{
spinwait2.spinonce();
}
result = this.m_array[i];
if (i + 1 >= 32)
{
spinwait2 = default(spinwait);
while (this.m_next == null)
{
spinwait2.spinonce();
}
head = this.m_next;
}
return true;
}
spinwait.spinonce();
i = this.low;
high = this.high;
}
result = default(t);
return false;
}
internal bool trypeek(out t result)
{
result = default(t);
int low = this.low;
if (low > this.high)
{
return false;
}
spinwait spinwait = default(spinwait);
while (this.m_state[low] == 0)
{
spinwait.spinonce();
}
result = this.m_array[low];
return true;
}
internal list<t> tolist(int start, int end)
{
list<t> list = new list<t>();
for (int i = start; i <= end; i++)
{
spinwait spinwait = default(spinwait);
while (this.m_state[i] == 0)
{
spinwait.spinonce();
}
list.add(this.m_array[i]);
}
return list;
}
}
上面的代码稍微分析一下就知道它的作用。concurrentqueue的线程安全的enqueue方法实现如下：
/// <summary>adds an object to the end of the <see cref="t:system.collections.concurrent.concurrentqueue`1" />.</summary>
/// <param name="item">the object to add to the end of the <see cref="t:system.collections.concurrent.concurrentqueue`1" />. the value can be a null reference (nothing in visual basic) for reference types.</param>
public void enqueue(t item)
{
spinwait spinwait = default(spinwait);
while (true)
{
concurrentqueue<t>.segment tail = this.m_tail;
if (tail.tryappend(item, ref this.m_tail))
{
break;
}
spinwait.spinonce();
}
}
concurrentqueue<t>线程安全实现过程小结：继承接口，子类内部通过同步构造实现接口的线程安全，直接对外公开调用。
和arraylist以及hashtable线程安全的“曲折”实现有点不同，concurrentqueue<t>一开始就是朝着线程安全方向实现去的。它没有使用lock，因为大家知道使用lock性能略差，对于读和写操作，应该分开，不能一概而论。concurrentqueue<t>具体实现在性能和异常处理上应该已经考虑的更全面周到一点。
在我看来，concurrentqueue<t>线程安全的具体实现有多吸引人在其次，iproducerconsumercollection<t>接口的抽象和提取非常值得称道，查看源码发现concurrentstack<t>和concurrentbag<t>也继承自该接口。<<clr via c#>>一书中在谈到接口和抽象类的时候特别举了集合和流(stream)的例子，微软为什么如此设计，想起来果然很有深意。
public bool tryadd(tkey key, tvalue value)
{
if (key == null)
{
throw new argumentnullexception("key");
}
tvalue tvalue;
return this.tryaddinternal(key, value, false, true, out tvalue);
}
其中内部方法tryadd的主要实现如下：
private bool tryaddinternal(tkey key, tvalue value, bool updateifexists, bool acquirelock, out tvalue resultingvalue)
{
checked
{
int hashcode = this.m_comparer.gethashcode(key);
concurrentdictionary<tkey, tvalue>.node[] buckets;
bool flag;
bool result;
while (true)
{
buckets = this.m_buckets;
int num;
int num2;
this.getbucketandlockno(hashcode, out num, out num2, buckets.length);
flag = false;
bool flag2 = false;
try
{
if (acquirelock)
{
monitor.enter(this.m_locks[num2], ref flag2);
}
if (buckets != this.m_buckets)
{
continue;
}
concurrentdictionary<tkey, tvalue>.node node = null;
for (concurrentdictionary<tkey, tvalue>.node node2 = buckets[num]; node2 != null; node2 = node2.m_next)
{
if (this.m_comparer.equals(node2.m_key, key))
{
if (updateifexists)
{
concurrentdictionary<tkey, tvalue>.node node3 = new concurrentdictionary<tkey, tvalue>.node(node2.m_key, value, hashcode, node2.m_next);
if (node == null)
{
buckets[num] = node3;
}
else
{
node.m_next = node3;
}
resultingvalue = value;
}
else
{
resultingvalue = node2.m_value;
}
result = false;
return result;
}
node = node2;
}
buckets[num] = new concurrentdictionary<tkey, tvalue>.node(key, value, hashcode, buckets[num]);
this.m_countperlock[num2]++;
if (this.m_countperlock[num2] > buckets.length / this.m_locks.length)
{
flag = true;
}
}
finally
{
if (flag2)
{
monitor.exit(this.m_locks[num2]);
}
}
break;
}
if (flag)
{
this.growtable(buckets);
goto il_131;
}
goto il_131;
return result;
il_131:
resultingvalue = value;
return true;
}
}
同步构造monitor瞬间吸引眼球，然后它的try…finally异常处理方式是不是也很眼熟？
2、concurrentdictionary<tkey, tvalue>
对于线程安全的泛型字典concurrentdictionary<tkey, tvalue>，我们也可以查看它的源码看它的具体实现方式。看源码有1200多行，实现稍微复杂一些。我们仅从最简单的tryadd方法分析：
四、如法炮制
如果让我来构造实现线程安全容器，最简单直接快速高效的方式就是参考arraylist和 hashtable，我们完全可以模仿它们的处理方式，通过继承一个容器，然后内部通过lock一个syncroot对象，中规中矩地实现framework中其他容器的线程安全。比如要实现线程安全的泛型队列queue<t>，贴一下大致的伪代码：
private class syncqueue<t> : queue<t>
{
#region fields and properties
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
}
通过类继承我们可以得到泛型队列的所有特点，需要实现线程安全的地方只要按需重写它即可，对外调用也很简单，直接模仿arraylist和hashtable，添加synchronized方法间接调用队列的子类即可，多么清晰简洁啊，关键时刻copy-paste也很有效嘛！
你可能觉得上面这样不动脑的方式似乎很傻很天真，但这绝对是一种正常人都能想到的思路，谁让ms的数组列表和哈希表就是这么实现的呢？
当然，我们还能想到的一种常见实现方式就是通过组合而不是类继承，实现的伪代码类似下面这样：
public class syncqueue<t>
{
#region fields and properties
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
}
上面这种方式和类继承的实现方式又有不同。它是通过在内部包装一个容器，然后按需进行方法、属性等等的线程安全处理，其他的所有特点依赖于那一个私有泛型队列组合对象queue。这种情况下泛型syncqueue和泛型队列是组合关系，依赖性和耦合性更低，相对更灵活，封装性更好，是一种较为通用的设计，实际开发和使用中这种方式比较常见。
到这里，我们至少可以分析得出，实现一般的线程安全容器的思路至少有两种：类继承（内部实现偏向使用组合）和（或）组合，线程安全的地方只要通过framework的同步构造如lock、interlocked等实现即可。
思考：如果让您实现线程安全容器，您优先会怎么实现呢？
五、线程安全并不真正安全
1、foreach遍历
cacheutil缓存实现的伪代码如下:
public class cacheutil
{
private static readonly hashtable ht = hashtable.synchronized(new hashtable());
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;
}
}
foreach的代码很简单，从哈希表构造的缓存中取数据并遍历，如下：
object obj = null;
var isok = cacheutil.tryget("key", out obj);
if (isok == false)
{
return;
}
var list = obj as ilist<t>;
if (list == null)
{
return;
}
foreach (var item in list) //遍历
{
//do something
}
上面的遍历代码一般情况下是不会有问题的。但是在多线程修改哈希表的value的情况下，上面的foreach遍历有可能发生异常。为什么呢？下面来简单分析一下：
从代码中可以看出来，哈希表中的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.
2、通过索引取集合中的数据
列表通过索引取值，一个简单的示例代码如下：
static int getfirstordefault(threadsafelist<int> list)
{
if (list.count > 0)
{
return list[0];
}
return 0;
}
当列表中的元素为1个的时候，上面的操作非常容易进入一个无厘头的陷阱之中。有人会问怎么会有陷阱，你看取数据之前都判断了啊，逻辑正确了啊，这哪里还有问题吗？
按照类似于1中的分析，getfirstordefault应该可以分为下面两步：
（1）线程1取数据，判断list.count的时候发现列表内有1个元素，这一步线程安全，没有任何问题，然后准备返回索引为0的元素；
（2）线程2在线程1将要取索引为0的元素之前移除了列表中的唯一元素或者直接将list指向null，这样线程1通过索引取元素就抛出异常了。
3、如何保证容器数据操作安全？
从上面的两个示例，我们得知通常所看到的线程安全实际上并不一定安全。不安全的主要原因就是容器内的数据很容易被其他线程改变，或者可以简要概括为：一段时间差引发的血案。实际上，我们平时所做的业务系统，归根结底很多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方法一般是没错的。