线程安全在高并发情况下是一个非常严重的问题。以下代码在多线程访问时，会出现问题。我们以List.Add为例，来说明在多线程访问下的状况。以下代码是List.Add的实现。
public void Add(T item) { if (this._size == this._items.Length) this.EnsureCapacity(this._size + 1); this._items[this._size++] = item; this._version++; }
当两个线程同时访问一个List的Add方法时，这个方法的第一条指令就可能出现不一致性了。因为，此时两个线程访问时_size都是一样的，正确情况下List应该执行EnsureCapacity(this._size + 2)而不再是EnsureCapacity(this._size + 1)了。为了确保List的线程安全，我们必须保证在任意时刻只能有一个线程来更改List的数据。这样我们的SimpleThreadSafeList就诞生了，在这个List中我们对其的每一个读写操作都加上一个排它锁。
public class SimpleThreadSafeList
: IList { private List _data; private object _syncRoot = new object(); public SimpleThreadSafeList() { _data = new List(); } public void RemoveAt(int index) { lock (_syncRoot) { _data.RemoveAt(index); } } public void Add(T item) { lock (_syncRoot) { _data.Add(item); } } // others...... }
这样我们确保List的数据在任意时刻都只有一个线程对其进行访问，看起来安全很多了。不过，如果所谓线程安全是这么简单的话，微软为什么不提供一个ThreadSafeList呢？JaredPar MSFT在这篇文章中做了一个描述《Why are thread safe collections so hard?http://blogs.msdn.com/b/jaredpar/archive/2009/02/11/why-are-thread-safe-collections-so-hard.aspx》，他解释微软不实现线程安全集合类，是因为以上这样的所谓“线程安全”并不是真正安全，我们可以从以下的代码中看出端倪。
var list = new SimpleThreadSafeList(); // ohters … if(list.Count > 0) { return list[0]; }
在以上代码中，我们创建了SimpleThreadSafeList这个类，其中有一个代码段就是用于获取list的默认值。如果有多个线程访问这段代码时，依然会出现数据不一致问题。即在执行“return list[0]”这条语句是，它是以“list.Count > 0”为前提的，当两个线程同时对这个SimpleThreadSafeList操作时，当前线程访问到list.Count大于0，但之后可能另一个线程将list清空了，这时候当前线程再来返回list[0]时就会出现IndexOutofRangeException了。SimpleThreadSafeList保证了List内部数据只能由一个线程来操作，但是对于上面的代码，它是无法保证数据不一致的。SimpleThreadSafeList仅能够被称为“数据线程安全”，这也是微软不提供线程安全集合类的原因了。JaredPar MSFT提出了一个真正解决线程安全的方法。那就是将SimpleThreadSafeList的_syncRoot 暴露出来。
public class SimpleThreadSafeList : IList { private object _syncRoot = new object(); public object SyncRoot { get { return _syncRoot; } } // others…… }
使用List时，需要使用到SyncRoot来加锁。
lock(list.SyncRoot) { if(list.Count > 0) { return list[0]; } }
不过，使用这种方式，有几个缺陷。第一，没有一个良好的Guide来指导编写线程安全的代码；第二，当SyncRoot使用范围过大时，非常容易造成死锁。下面是一段可能产生死锁的代码。
var list1 = new SimpleThreadSafeList();
var list2 = new SimpleThreadSafeList(); new Thread(()=> { lock(list1.SyncRoot) { list2.Add(10); //间接请求了list2.SyncRoot } }).Start(); new Thread(()=> { lock(list2.SyncRoot) { list1.Add(10); //间接请求了list1.SyncRoot } }).Start();
对于死锁这个问题，我们采取的方法是使用Monitor.TryEnter，从而来避免一直死锁。对于前一个问题，我这边仅是基于DisposableLocker来实现尽可能的线程安全，对于如何使用，我目前依然没有一个良好的理论，只能说我们在设计高并发的API时，对多个线程可以同时访问的对象都需要加以判断从而来确定需要采用什么样的方式处理。
namespace UIShell.OSGi.Collection { /// /// Use Monitor.TryEnter to acquire a lock. This would not cause the dead lock. /// public class DisposableLocker : IDisposable { private object _syncRoot; private bool _lockAcquired; private int _millisecondsTimeout; public DisposableLocker(object syncRoot, int millisecondsTimeout) { _syncRoot = syncRoot; _millisecondsTimeout = millisecondsTimeout; _lockAcquired = Monitor.TryEnter(_syncRoot, _millisecondsTimeout); LogWhenAcquireLockFailed(); } private void LogWhenAcquireLockFailed() { if (!_lockAcquired) //这时候可能要如下处理：（1）记录日志；（2）记录日志并抛出异常；（3）记录日志，然后Block，重现死锁。 { FileLogUtility.Error(string.Format( "Accquire the lock timeout. The limited time is {0} milliseconds.", _millisecondsTimeout )); } } public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } private void Dispose(bool disposing) { if (disposing) { if (_lockAcquired) { Monitor.Exit(_syncRoot); } } } ~DisposableLocker() { Dispose(false); } } }
当出现死锁时，这里采用的解决方案是记录日志，然后继续运行。不过，这种方法有一定缺陷，可能在极端情况下引起数据不一致。因此，我们可能需要抛出异常或者让锁一直持续下去。SimpleThreadSafeList此时将以以下方式来实现。
public class SimpleThreadSafeList : IList { private object _syncRoot = new object(); public object SyncRoot { get { return _syncRoot; } } public int MillisecondsTimeoutOnLock { get; private set; } public DisposableLocker CreateLocker() { return new DisposableLocker(MillisecondsTimeoutOnLock, SyncRoot); } public void Add(T item) { using(CreateLocker()) { _data.Add(item); } } // others…… }
接下来使用List.CreateLocker来建立一个全局锁
using(list.CreateLocker()) { if(list.Count > 0) { return list[0]; } }
大家可以看一下是否还有更好的方式来保证线程绝对安全。
延伸阅读
最新评论