问题的提出
所谓单个写入程序/多个阅读程序的线程同步问题是指任意数量的线程访问共享资源时写入程序（线程）需要修改共享资源而阅读程序（线程）需要读取数据在这个同步问题中很容易得到下面二个要求
）当一个线程正在写入数据时其他线程不能写也不能读
）当一个线程正在读入数据时其他线程不能写但能够读
在数据库应用程序环境中经常遇到这样的问题比如说有n个最终用户他们都要同时访问同一个数据库其中有m个用户要将数据存入数据库nm个用户要读取数据库中的记录
很显然在这个环境中我们不能让两个或两个以上的用户同时更新同一条记录如果两个或两个以上的用户都试图同时修改同一记录那么该记录中的信息就会被破坏
我们也不让一个用户更新数据库记录的同时让另一用户读取记录的内容因为读取的记录很有可能同时包含了更新和没有更新的信息也就是说这条记录是无效的记录
实现分析
规定任一线程要对资源进行写或读操作前必须申请锁根据操作的不同分为阅读锁和写入锁操作完成之后应释放相应的锁将单个写入程序/多个阅读程序的要求改变一下可以得到如下的形式
一个线程申请阅读锁的成功条件是当前没有活动的写入线程
一个线程申请写入锁的成功条件是当前没有任何活动（对锁而言）的线程
因此为了标志是否有活动的线程以及是写入还是阅读线程引入一个变量m_nActive如果m_nActive > 则表示当前活动阅读线程的数目如果m_nActive=则表示没有任何活动线程m_nActive <表示当前有写入线程在活动注意m_nActive<时只能取的值因为只允许有一个写入线程活动
为了判断当前活动线程拥有的锁的类型我们采用了线程局部存储技术（请参阅其它参考书籍）将线程与特殊标志位关联起来
申请阅读锁的函数原型为public void AcquireReaderLock( int millisecondsTimeout )其中的参数为线程等待调度的时间函数定义如下
public void AcquireReaderLock( int millisecondsTimeout ) { // m_mutext很快可以得到以便进入临界区 m_mutexWaitOne( ); // 是否有写入线程存在 bool bExistingWriter = ( m_nActive < ); if( bExistingWriter ) { //等待阅读线程数目加当有锁释放时根据此数目来调度线程 m_nWaitingReaders++; } else { //当前活动线程加 m_nActive++; } m_mutexReleaseMutex(); //存储锁标志为Reader SystemLocalDataStoreSlot slot = ThreadGetNamedDataSlot(m_strThreadSlotName); object obj = ThreadGetData( slot ); LockFlags flag = LockFlagsNone; if( obj != null ) flag = (LockFlags)obj ; if( flag == LockFlagsNone ) { ThreadSetData( slot LockFlagsReader ); } else { ThreadSetData( slot (LockFlags)((int)flag | (int)LockFlagsReader ) ); } if( bExistingWriter ) { //等待指定的时间 thism_aeReadersWaitOne( millisecondsTimeout true ); } }
它首先进入临界区（用以在多线程环境下保证活动线程数目的操作的正确性）判断当前活动线程的数目如果有写线程(m_nActive<)存在则等待指定的时间并且等待的阅读线程数目加如果当前活动线程是读线程(m_nActive>=)则可以让读线程继续运行
申请写入锁的函数原型为public void AcquireWriterLock( int millisecondsTimeout )其中的参数为等待调度的时间函数定义如下
public void AcquireWriterLock( int millisecondsTimeout ) { // m_mutext很快可以得到以便进入临界区 m_mutexWaitOne( ); // 是否有活动线程存在 bool bNoActive = m_nActive == ; if( !bNoActive ) { m_nWaitingWriters++; } else { m_nActive; } m_mutexReleaseMutex(); //存储线程锁标志 SystemLocalDataStoreSlot slot = ThreadGetNamedDataSlot( myReaderWriterLockDataSlot ); object obj = ThreadGetData( slot ); LockFlags flag = LockFlagsNone; if( obj != null ) flag = (LockFlags)ThreadGetData( slot ); if( flag == LockFlagsNone ) { ThreadSetData( slot LockFlagsWriter ); } else { ThreadSetData( slot (LockFlags)((int)flag | (int)LockFlagsWriter ) ); } //如果有活动线程等待指定的时间 if( !bNoActive ) thism_aeWritersWaitOne( millisecondsTimeout true ); }
它首先进入临界区判断当前活动线程的数目如果当前有活动线程存在不管是写线程还是读线程（m_nActive）线程将等待指定的时间并且等待的写入线程数目加否则线程拥有写的权限
释放阅读锁的函数原型为public void ReleaseReaderLock()函数定义如下
public void ReleaseReaderLock() { SystemLocalDataStoreSlot slot = ThreadGetNamedDataSlot(m_strThreadSlotName ); LockFlags flag = (LockFlags)ThreadGetData( slot ); if( flag == LockFlagsNone ) { return; } bool bReader = true; switch( flag ) { case LockFlagsNone: break; case LockFlagsWriter: bReader = false; break; } if( !bReader ) return; ThreadSetData( slot LockFlagsNone ); m_mutexWaitOne(); AutoResetEvent autoresetevent = null; thism_nActive ; if( thism_nActive == ) { if( thism_nWaitingReaders > ) { m_nActive ++ ; m_nWaitingReaders ; autoresetevent = thism_aeReaders; } else if( thism_nWaitingWriters > ) { m_nWaitingWriters; m_nActive ; autoresetevent = thism_aeWriters ; } } m_mutexReleaseMutex(); if( autoresetevent != null ) autoreseteventSet(); }
释放阅读锁时首先判断当前线程是否拥有阅读锁（通过线程局部存储的标志）然后判断是否有等待的阅读线程如果有先将当前活动线程加等待阅读线程数目减然后置事件为有信号如果没有等待的阅读线程判断是否有等待的写入线程如果有则活动线程数目减等待的写入线程数目减释放写入锁与释放阅读锁的过程基本一致可以参看源代码
注意在程序中释放锁时只会唤醒一个阅读程序这是因为使用AutoResetEvent的原历读者可自行将其改成ManualResetEvent同时唤醒多个阅读程序此时应令m_nActive等于整个等待的阅读线程数目
测试
测试程序取自Net FrameSDK中的一个例子只是稍做修改测试程序如下
using System; using SystemThreading; using MyThreading; class Resource { myReaderWriterLock rwl = new myReaderWriterLock(); public void Read(Int threadNum) { rwlAcquireReaderLock(TimeoutInfinite); try { ConsoleWriteLine(Start Resource reading (Thread={}) threadNum); ThreadSleep(); ConsoleWriteLine(Stop Resource reading (Thread={}) threadNum); } finally { rwlReleaseReaderLock(); } } public void Write(Int threadNum) { rwlAcquireWriterLock(TimeoutInfinite); try { ConsoleWriteLine(Start Resource writing (Thread={}) threadNum); ThreadSleep(); ConsoleWriteLine(Stop Resource writing (Thread={}) threadNum); } finally { rwlReleaseWriterLock(); } } } class App { static Int numAsyncOps = ; static AutoResetEvent asyncOpsAreDone = new AutoResetEvent(false); static Resource res = new Resource(); public static void Main() { for (Int threadNum = ; threadNum < ; threadNum++) { ThreadPoolQueueUserWorkItem(new WaitCallback(UpdateResource) threadNum);} asyncOpsAreDoneWaitOne(); ConsoleWriteLine(All operations have completed); ConsoleReadLine(); } // The callback methods signature MUST match that of a SystemThreadingTimerCallback // delegate (it takes an Object parameter and returns void) static void UpdateResource(Object state) { Int threadNum = (Int) state; if ((threadNum % ) != ) resRead(threadNum); else resWrite(threadNum); if (InterlockedDecrement(ref numAsyncOps) == ) asyncOpsAreDoneSet(); } }
