如何处理几十万条并发数据

62 阅读 0 评论 41 点赞

我是靠谱客的博主聪明小天鹅，最近开发中收集的这篇文章主要介绍如何处理几十万条并发数据，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

我在博客堂上也看到不少有关并发控制的文章，我一直是推荐使用时间戳来解决的。
比如我们在SQL Server中的表中定义一个字段为timestamp类型的字段ts,这个字段的值不需要我们进行控制的。
在Insert与Update时，数据库会自己进行ts值的更新，因此我们只要在Update时使用：
Update xxx where key=@key and ts=@ts 就可以了，根本不用考虑ts里的值更新。

Delete时也最好进行一下判断，用这种方式是可以控制数据并发操作的。
只需要在Update与Delete时，判断"影响条数"就可以知道更新是否成功。

这一点我想非常方便，但不是所有的数据库都支持timestampr的，如在Access里没有timestamp字段，我也不知道其他的数据库是否都有类似的timestamp类型，不管怎么样，我觉得我们不能完全信赖于数据库的控制，而应该采用自设的控制机制，这样可以适应系统的数据库移值，下面我就介绍一下，在.NET下如何实现，自设的时间戳控制。

我们也同样建一个字段ts，定义为Varchar，长度在20以上即可，而且不允许为null，这样比较合适。

我们应该采用什么机制来生成随时的或者说不可能会产生一样的值，我推荐的是DateTime.Now.Ticks，这是一个12位的数字，由于在Update等更新时，数据库会自动进行锁定，所以不可能会在同一时间会有两个一样的操作执行，因此这就可以避免Ticks产生相同的值了。

或者也可以采用Guid值，也可以产生唯一值，但我觉得Guid值太大，可能会影响效率。

那好，在我们Insert时:Insert xxxx ts='221283747584' where key='1'
在Update时 Update xx set xxx ts='39383848593839' where key='1' and ts='111111111111' //假设取到的原值为'11111111111'
Delete类似上面的。

我们判断影响条数，如果为0则说明更新不成功。

我相信上面的方法是简单可行的。

我目前也遇到一个问题：如果是批量更新与批量删除，如何进行并发控制呢？
由于批量更新时，不是一条记录：Update xxx where Birthday='2004-2-1'之类的，会影响到N条数据，要进行并发控制就不那么容易了。如果还是采用一条条判断ts那是不现实的。
对于这种只能放弃并发控制吗？

在多线程编程时，开发人员经常会遭遇多个线程读写某个资源的情况。这就需要进行线程同步来保证线程安全。一般情况下，我们的同步措施是使用锁机制。但是，假如线程只对资源进行读取操作，那么根本不需要使用锁；反之，假如线程只对资源进行写入操作，则应当使用互斥锁(比如使用 Monitor类等)。还有一种情况，就是存在多个线程对资源进行读取操作，同时每次只有一个线程对资源进行独占写入操作。这正是本文主题--读写锁的用武之地。ReaderWriterLock 类

　　.NET Framework BCL 在 1.1 版本时，给我们提供了一个ReaderWriterLock 类来面对此种情景。但是很遗憾，Microsoft 官方不推荐使用该类。Jeffrey Richter也在他的《CLR via C#》一书中对它进行了严厉的批判。下面是该类不受欢迎的主要原因：

　　性能。这个类实在是太慢了。比如它的 AcquireReaderLock 方法比 Monitor 类的 Enter 方法要慢 5 倍左右，而等待争夺写锁甚至比 Monitor 类慢 6 倍。

　　策略。假如某个线程完成写入操作后，同时面临读线程和写线程等待处理。ReaderWriterLock会优先释放读线程，而让写线程继续等待。但我们使用读写锁是因为存在大量的读线程和非常少的写线程，这样写线程很可能必须长时间地等待，造成写线程饥饿，不能及时更新数据。更槽糕的情况是，假如写线程一直等待，就会造成活锁。反之，我们让 ReaderWriterLock采取写线程优先的策略。如果存在多个写线程，而读线程数量稀少，也会造成读线程饥饿。幸运的是，现实实践中，这种情况很少出现。一旦发生这种情况，我们可以采取互斥锁的办法。

递归。ReaderWriterLock类支持锁递归。这就意味着该锁清楚的知道目前哪个线程拥有它。假如拥有该锁的线程递归尝试获得该读写锁，递归算法允许该线程获得该读写锁，并且增加获得该锁的计数。然而该线程必须释放该锁相同的次数以便线程不再拥有该锁。尽管这看起来是个很好的特性，但是实现这个“特性”代价太高。首先，因为多个读线程可以同时拥有该读写锁，这必须让该锁为每个线程保持计数。此外，还需要额外的内存空间和时间来更新计数。这个特性对 ReaderWriterLock类可怜的性能贡献极大。其次，有些良好的设计需要一个线程在此处获得该锁，然后在别处释放该锁(比如 .NET 的异步编程架构)。因为这个递归特性，ReaderWriterLock 不支持这种编程架构。

　　资源泄漏。在 .NET 2.0 之前的版本中， ReaderWriterLock 类会造成内核对象泄露。这些对象只有在进程终止后才能再次回收。幸运的是，.NET 2.0 修正了这个 Bug 。

　　此外，ReaderWriterLock 还有个令人担心的危险的非原子性操作。它就是 UpgradeToWriteLock方法。这个方法实际上在更新到写锁前先释放了读锁。这就让其他线程有机会在此期间乘虚而入，从而获得读写锁且改变状态。如果先更新到写锁，然后释放读锁。假如两个线程同时更新将会导致另外一个线程死锁。

　　所以 Microsoft 决定构建一个新类来一次性解决上述所有问题，这就是ReaderWriterLockSlim 类。本来可以在原有的 ReaderWriterLock 类上修正错误，但是考虑到兼容性和已存在的API ，Microsoft 放弃了这种做法。当然也可以标记 ReaderWriterLock 类为Obsolete，但是由于某些原因，这个类还有存在的必要。

　　ReaderWriterLockSlim 类

　　新的ReaderWriterLockSlim 类支持三种锁定模式：Read，Write，UpgradeableRead。这三种模式对应的方法分别是EnterReadLock，EnterWriteLock，EnterUpgradeableReadLock 。再就是与此对应的TryEnterReadLock，TryEnterWriteLock，TryEnterUpgradeableReadLock，ExitReadLock，ExitWriteLock，ExitUpgradeableReadLock。Read 和 Writer 锁定模式比较简单易懂：Read 模式是典型的共享锁定模式，任意数量的线程都可以在该模式下同时获得锁；Writer模式则是互斥模式，在该模式下只允许一个线程进入该锁。UpgradeableRead锁定模式可能对于大多数人来说比较新鲜，但是在数据库领域却众所周知。

　　这个新的读写锁类性能跟 Monitor 类大致相当，大概在Monitor 类的 2倍之内。而且新锁优先让写线程获得锁，因为写操作的频率远小于读操作。通常这会导致更好的可伸缩性。起初，ReaderWriterLockSlim类在设计时考虑到相当多的情况。比如在早期 CTP 的代码还提供了PrefersReaders，PrefersWritersAndUpgrades 和 Fifo 等竞争策略。但是这些策略虽然添加起来非常简单，但是会导致情况非常的复杂。所以Microsoft 最后决定提供一个能够在大多数情况下良好工作的简单模型。

　　ReaderWriterLockSlim 的更新锁

　　现在让我们更加深入的讨论一下更新模型。UpgradeableRead 锁定模式允许安全的从 Read 或 Write 模式下更新。还记得先前ReaderWriterLock的更新是非原子性，危险的操作吗(尤其是大多数人根本没有意识到这点)？现在提供的新读写锁既不会破坏原子性，也不会导致死锁。新锁一次只允许一个线程处在 UpgradeableRead 模式下。

　　一旦该读写锁处在 UpgradeableRead模式下，线程就能读取某些状态值来决定是否降级到 Read 模式或升级到 Write 模式。注意应当尽可能快的作出这个决定：持有UpgradeableRead 锁会强制任何新的读请求等待，尽管已存在的读取操作仍然活跃。遗憾的是，CLR 团队移除了DowngradeToRead 和 UpgradeToWrite 两个方法。如果要降级到读锁，只要简单的在ExitUpgradeableReadLock 方法后紧跟着调用 EnterReadLock 方法即可：这可以让其他的 Read 和UpgradeableRead 获得完成先前应当持有却被 UpgradeableRead 锁持有的操作。如果要升级到写锁，只要简单调用EnterWriteLock 方法即可：这可能要等待，直到不再有任何线程在 Read 模式下持有锁。不像降级到读锁，必须调用ExitUpgradeableReadLock。在 Write 模式下不必非得调用ExitUpgradeableReadLock。但是为了形式统一，最好还是调用它。比如下面的代码：

using System; using System.Linq; using System.Threading; 　　 namespace Lucifer.CSharp.Sample { 　　class Program 　　{ 　　　　private ReaderWriterLockSlim rwLock = new ReaderWriterLockSlim(); 　　　　　　void Sample() 　　　　{ 　　　　　　bool isUpdated = true; 　　　　　　rwLock.EnterUpgradeableReadLock(); 　　　　　　try 　　　　　　{ 　　　　　　　　if (/* … 读取状态值来决定是否更新 … */) 　　　　　　　　{ 　　　　　　　　　　rwLock.EnterWriteLock(); 　　　　　　　　　　try 　　　　　　　　　　{ 　　　　　　　　　　　　//… 写入状态值 … 　　　　　　　　　　} 　　　　　　　　　　finally 　　　　　　　　　　{ 　　　　　　　　　　　　rwLock.ExitWriteLock(); 　　　　　　　　　　} 　　　　　　　　} 　　　　　　　　else 　　　　　　　　{ 　　　　　　　　　　rwLock.EnterReadLock(); 　　　　　　　　　　rwLock.ExitUpgradeableReadLock(); 　　　　　　　　　　isUpdated = false; 　　　　　　　　　　try 　　　　　　　　　　{ 　　　　　　　　　　　　//… 读取状态值 … 　　　　　　　　　　} 　　　　　　　　　　finally 　　　　　　　　　　{ 　　　　　　　　　　　　rwLock.ExitReadLock(); 　　　　　　　　　　} 　　　　　　　　} 　　　　　　} 　　　　　　finally 　　　　　　{ 　　　　　　　　if (isUpdated) 　　　　　　　　　　rwLock.ExitUpgradeableReadLock(); 　　　　　　} 　　　　} 　　} }ReaderWriterLockSlim 的递归策略

　　新的读写锁还有一个有意思的特性就是它的递归策略。默认情况下，除已提及的降级到读锁和升级到写锁之外，所有的递归请求都不允许。这意味着你不能连续两次调用 EnterReadLock，其他模式下也类似。如果你这么做，CLR 将会抛出 LockRecursionException异常。当然，你可以使用 LockRecursionPolicy.SupportsRecursion的构造函数参数让该读写锁支持递归锁定。但不建议对新的开发使用递归，因为递归会带来不必要的复杂情况，从而使你的代码更容易出现死锁现象。

　　有一种特殊的情况永远也不被允许，无论你采取什么样的递归策略。这就是当线程持有读锁时请求写锁。Microsoft 曾经考虑提供这样的支持，但是这种情况太容易导致死锁。所以 Microsoft 最终放弃了这个方案。

　　此外，这个新的读写锁还提供了很多对应的属性来确定线程是否在指定模型下持有该锁。比如 IsReadLockHeld，IsWriteLockHeld 和 IsUpgradeableReadLockHeld 。你也可以通过WaitingReadCount，WaitingWriteCount 和 WaitingUpgradeCount等属性来查看有多少线程正在等待持有特定模式下的锁。CurrentReadCount属性则告知目前有多少并发读线程。RecursiveReadCount， RecursiveWriteCount 和RecursiveUpgradeCount 则告知目前线程进入特定模式锁定状态下的次数。

　　小结

　　这篇文章分析了.NET 中提供的两个读写锁类。然而 .NET 3.5 提供的新读写锁 ReaderWriterLockSlim 类消除了ReaderWriterLock 类存在的主要问题。与 ReaderWriterLock相比，性能有了极大提高。更新具有原子性，也可以极大避免死锁。更有清晰的递归策略。在任何情况下，我们都应该使用ReaderWriterLockSlim 来代替 ReaderWriterLock 类。

　　Update 于 2008-12-07 0:06

Windows Vista 及其以后的版本新增了一个 SRWLock 原语。它以 Windows 内核事件机制为基础而构建。它的设计比较有意思。

　　SRW 锁不支持递归。Windows Kernel 团队认为支持递归会造成额外系统开销，原因是为了维持准确性需进行逐线程的计数。SRW锁也不支持从共享访问升级到独占访问，同时也不支持从独占访问降级到共享访问。支持升级能力可能会造成难以接受的复杂性和额外系统开销，这种开销甚至会影响锁内共享和独占获得代码的常见情况。它还要求定义关于如何选择等待中的读取器、等待中的写入器和等待升级的读取器的策略，这又将与无偏向的基本设计目标相抵触。我对其进行了 .NET 封装。代码如下：

using System; using System.Threading; using System.Runtime.InteropServices; 　　 namespace Lucifer.Threading.Lock { 　　/// <summary> 　　/// Windows NT 6.0 才支持的读写锁。　　/// </summary> 　　/// <remarks>请注意，这个类只能在 NT 6.0 及以后的版本中才能使用。</remarks> 　　public sealed class SRWLock 　　{ 　　　　private IntPtr rwLock; 　　　　　　/// <summary> 　　　　/// 该锁不支持递归。　　　　/// </summary> 　　　　public SRWLock() 　　　　{ 　　　　　　InitializeSRWLock(out rwLock); 　　　　} 　　　　　　/// <summary> 　　　　/// 获得读锁。　　　　/// </summary> 　　　　public void EnterReadLock() 　　　　{ 　　　　　　AcquireSRWLockShared(ref rwLock); 　　　　} 　　　　　　/// <summary> 　　　　/// 获得写锁。　　　　/// </summary> 　　　　public void EnterWriteLock() 　　　　{ 　　　　　　AcquireSRWLockExclusive(ref rwLock); 　　　　} 　　　　　　/// <summary> 　　　　/// 释放读锁。　　　　/// </summary> 　　　　public void ExitReadLock() 　　　　{ 　　　　　　ReleaseSRWLockShared(ref rwLock); 　　　　} 　　　　　　/// <summary> 　　　　/// 释放写锁。　　　　/// </summary> 　　　　public void ExitWriteLock() 　　　　{ 　　　　　　ReleaseSRWLockExclusive(ref rwLock); 　　　　} 　　　　　　[DllImport("Kernel32", CallingConvention = CallingConvention.Winapi, ExactSpelling = true)] 　　　　private static extern void InitializeSRWLock(out IntPtr rwLock); 　　　　　　[DllImport("Kernel32", CallingConvention = CallingConvention.Winapi, ExactSpelling = true)] 　　　　private static extern void AcquireSRWLockExclusive(ref IntPtr rwLock); 　　　　　　[DllImport("Kernel32", CallingConvention = CallingConvention.Winapi, ExactSpelling = true)] 　　　　private static extern void AcquireSRWLockShared(ref IntPtr rwLock); 　　　　　　[DllImport("Kernel32", CallingConvention = CallingConvention.Winapi, ExactSpelling = true)] 　　　　private static extern void ReleaseSRWLockExclusive(ref IntPtr rwLock); 　　　　　　[DllImport("Kernel32", CallingConvention = CallingConvention.Winapi, ExactSpelling = true)] 　　　　private static extern void ReleaseSRWLockShared(ref IntPtr rwLock); 　　} }此外，在其他平台也有一些有意思的读写锁。比如 Linux 内核中的读写锁和 Java 中的读写锁。感兴趣的同学可以自己研究一番。

读写锁有个很常用的场景就是在缓存设计中。因为缓存中经常有些很稳定，不太长更新的内容。MSDN 的代码示例就很经典，我原版拷贝一下，呵呵。代码示例如下：

using System; using System.Threading; 　　 namespace Lucifer.CSharp.Sample { 　　public class SynchronizedCache 　　{ 　　　　private ReaderWriterLockSlim cacheLock = new ReaderWriterLockSlim(); 　　　　private Dictionary<int, string> innerCache = new Dictionary<int, string>(); 　　　　　　public string Read(int key) 　　　　{ 　　　　　　cacheLock.EnterReadLock(); 　　　　　　try 　　　　　　{ 　　　　　　　　return innerCache[key]; 　　　　　　} 　　　　　　finally 　　　　　　{ 　　　　　　　　cacheLock.ExitReadLock(); 　　　　　　} 　　　　} 　　　　　　public void Add(int key, string value) 　　　　{ 　　　　　　cacheLock.EnterWriteLock(); 　　　　　　try 　　　　　　{ 　　　　　　　　innerCache.Add(key, value); 　　　　　　} 　　　　　　finally 　　　　　　{ 　　　　　　　　cacheLock.ExitWriteLock(); 　　　　　　} 　　　　} 　　　　　　public bool AddWithTimeout(int key, string value, int timeout) 　　　　{ 　　　　　　if (cacheLock.TryEnterWriteLock(timeout)) 　　　　　　{ 　　　　　　　　try 　　　　　　　　{ 　　　　　　　　　　innerCache.Add(key, value); 　　　　　　　　} 　　　　　　　　finally 　　　　　　　　{ 　　　　　　　　　　cacheLock.ExitWriteLock(); 　　　　　　　　} 　　　　　　　　return true; 　　　　　　} 　　　　　　else 　　　　　　{ 　　　　　　　　return false; 　　　　　　} 　　　　} 　　　　　　public AddOrUpdateStatus AddOrUpdate(int key, string value) 　　　　{ 　　　　　　cacheLock.EnterUpgradeableReadLock(); 　　　　　　try 　　　　　　{ 　　　　　　　　string result = null; 　　　　　　　　if (innerCache.TryGetValue(key, out result)) 　　　　　　　　{ 　　　　　　　　　　if (result == value) 　　　　　　　　　　{ 　　　　　　　　　　　　return AddOrUpdateStatus.Unchanged; 　　　　　　　　　　} 　　　　　　　　　　else 　　　　　　　　　　{ 　　　　　　　　　　　　cacheLock.EnterWriteLock(); 　　　　　　　　　　　　try 　　　　　　　　　　　　{ 　　　　　　　　　　　　　　innerCache[key] = value; 　　　　　　　　　　　　} 　　　　　　　　　　　　finally 　　　　　　　　　　　　{ 　　　　　　　　　　　　　　cacheLock.ExitWriteLock(); 　　　　　　　　　　　　} 　　　　　　　　　　　　return AddOrUpdateStatus.Updated; 　　　　　　　　　　} 　　　　　　　　} 　　　　　　　　else 　　　　　　　　{ 　　　　　　　　　　cacheLock.EnterWriteLock(); 　　　　　　　　　　try 　　　　　　　　　　{ 　　　　　　　　　　　　innerCache.Add(key, value); 　　　　　　　　　　} 　　　　　　　　　　finally 　　　　　　　　　　{ 　　　　　　　　　　　　cacheLock.ExitWriteLock(); 　　　　　　　　　　} 　　　　　　　　　　return AddOrUpdateStatus.Added; 　　　　　　　　} 　　　　　　} 　　　　　　finally 　　　　　　{ 　　　　　　　　cacheLock.ExitUpgradeableReadLock(); 　　　　　　} 　　　　} 　　　　　　public void Delete(int key) 　　　　{ 　　　　　　cacheLock.EnterWriteLock(); 　　　　　　try 　　　　　　{ 　　　　　　　　innerCache.Remove(key); 　　　　　　} 　　　　　　finally 　　　　　　{ 　　　　　　　　cacheLock.ExitWriteLock(); 　　　　　　} 　　　　} 　　　　　　public enum AddOrUpdateStatus 　　　　{ 　　　　　　Added, 　　　　　　Updated, 　　　　　　Unchanged 　　　　}; 　　} }