Java并发编程--ReentrantReadWriteLock

 1 class CachedData {
 2     Object data;
 3     volatile boolean cacheValid;    //缓存是否有效
 4     ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
 5 
 6     void processCachedData() {
 7         rwl.readLock().lock();    //获取读锁
 8         //如果缓存无效，更新cache;否则直接使用data
 9         if (!cacheValid) {
10             // Must release read lock before acquiring write lock
11             //获取写锁前须释放读锁
12             rwl.readLock().unlock();
13             rwl.writeLock().lock();    
14             // Recheck state because another thread might have acquired
15             //   write lock and changed state before we did.
16             if (!cacheValid) {
17                 data = ...
18                 cacheValid = true;
19             }
20             // Downgrade by acquiring read lock before releasing write lock
21             //锁降级，在释放写锁前获取读锁
22             rwl.readLock().lock();
23             rwl.writeLock().unlock(); // Unlock write, still hold read
24         }
25 
26         use(data);
27         rwl.readLock().unlock();    //释放读锁
28     }
29 }

 1 class RWDictionary {
 2     private final Map<String, Data> m = new TreeMap<String, Data>();
 3     private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
 4     private final Lock r = rwl.readLock();    //读锁
 5     private final Lock w = rwl.writeLock();    //写锁
 6 
 7     public Data get(String key) {
 8         r.lock();
 9         try { return m.get(key); }
10         finally { r.unlock(); }
11     }
12     public String[] allKeys() {
13         r.lock();
14         try { return m.keySet().toArray(); }
15         finally { r.unlock(); }
16     }
17     public Data put(String key, Data value) {
18         w.lock();
19         try { return m.put(key, value); }
20         finally { w.unlock(); }
21     }
22     public void clear() {
23         w.lock();
24         try { m.clear(); }
25         finally { w.unlock(); }
26     }
27 }

1 /** Inner class providing readlock */
2 private final ReentrantReadWriteLock.ReadLock readerLock;
3 /** Inner class providing writelock */
4 private final ReentrantReadWriteLock.WriteLock writerLock;
5 /** Performs all synchronization mechanics */
6 final Sync sync;

 1 //获取写锁
 2 public void lock() {
 3     sync.acquire(1);
 4 }
 5 
 6 //AQS实现的独占式获取同步状态方法
 7 public final void acquire(int arg) {
 8     if (!tryAcquire(arg) &&
 9         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
10         selfInterrupt();
11 }
12 
13 //自定义重写的tryAcquire方法
14 protected final boolean tryAcquire(int acquires) {
15     /*
16      * Walkthrough:
17      * 1. If read count nonzero or write count nonzero
18      *    and owner is a different thread, fail.
19      * 2. If count would saturate, fail. (This can only
20      *    happen if count is already nonzero.)
21      * 3. Otherwise, this thread is eligible for lock if
22      *    it is either a reentrant acquire or
23      *    queue policy allows it. If so, update state
24      *    and set owner.
25      */
26     Thread current = Thread.currentThread();
27     int c = getState();
28     int w = exclusiveCount(c);    //取同步状态state的低16位，写同步状态
29     if (c != 0) {
30         // (Note: if c != 0 and w == 0 then shared count != 0)
31         //存在读锁或当前线程不是已获取写锁的线程，返回false
32         if (w == 0 || current != getExclusiveOwnerThread())
33             return false;
34         //判断同一线程获取写锁是否超过最大次数，支持可重入
35         if (w + exclusiveCount(acquires) > MAX_COUNT)    //
36             throw new Error("Maximum lock count exceeded");
37         // Reentrant acquire
38         setState(c + acquires);
39         return true;
40     }
41     //此时c=0,读锁和写锁都没有被获取
42     if (writerShouldBlock() ||
43         !compareAndSetState(c, c + acquires))
44         return false;
45     setExclusiveOwnerThread(current);
46     return true;
47 }

1 //FairSync中需要判断是否有前驱节点，如果有则返回false，否则返回true。遵循FIFO
2 final boolean writerShouldBlock() {
3     return hasQueuedPredecessors();
4 }
5 
6 //NonfairSync中直接返回false，可插队。
7 final boolean writerShouldBlock() {
8     return false; // writers can always barge
9 }

 1 //写锁释放
 2 public void unlock() {
 3     sync.release(1);
 4 }
 5 
 6 //AQS提供独占式释放同步状态的方法
 7 public final boolean release(int arg) {
 8     if (tryRelease(arg)) {
 9         Node h = head;
10         if (h != null && h.waitStatus != 0)
11             unparkSuccessor(h);
12         return true;
13     }
14     return false;
15 }
16 
17 //自定义重写的tryRelease方法
18 protected final boolean tryRelease(int releases) {
19     if (!isHeldExclusively())
20         throw new IllegalMonitorStateException();
21     int nextc = getState() - releases;    //同步状态减去releases
22     //判断同步状态的低16位（写同步状态）是否为0，如果为0则返回true，否则返回false.
23     //因为支持可重入
24     boolean free = exclusiveCount(nextc) == 0;    
25     if (free)
26         setExclusiveOwnerThread(null);
27     setState(nextc);    //以获取写锁，不需要其他同步措施，是线程安全的
28     return free;
29 }

 1 public void lock() {
 2     sync.acquireShared(1);
 3 }
 4 
 5 //使用AQS提供的共享式获取同步状态的方法
 6 public final void acquireShared(int arg) {
 7     if (tryAcquireShared(arg) < 0)
 8         doAcquireShared(arg);
 9 }
10 
11 //自定义重写的tryAcquireShared方法，参数是unused，因为读锁的重入计数是内部维护的
12 protected final int tryAcquireShared(int unused) {
13     /*
14      * Walkthrough:
15      * 1. If write lock held by another thread, fail.
16      * 2. Otherwise, this thread is eligible for
17      *    lock wrt state, so ask if it should block
18      *    because of queue policy. If not, try
19      *    to grant by CASing state and updating count.
20      *    Note that step does not check for reentrant
21      *    acquires, which is postponed to full version
22      *    to avoid having to check hold count in
23      *    the more typical non-reentrant case.
24      * 3. If step 2 fails either because thread
25      *    apparently not eligible or CAS fails or count
26      *    saturated, chain to version with full retry loop.
27      */
28     Thread current = Thread.currentThread();
29     int c = getState();
30     //exclusiveCount(c)取低16位写锁。存在写锁且当前线程不是获取写锁的线程，返回-1，获取读锁失败。
31     if (exclusiveCount(c) != 0 &&
32         getExclusiveOwnerThread() != current)
33         return -1;
34     int r = sharedCount(c);    //取高16位读锁，
35     //readerShouldBlock（）用来判断当前线程是否应该被阻塞
36     if (!readerShouldBlock() &&
37         r < MAX_COUNT &&    //MAX_COUNT为获取读锁的最大数量，为16位的最大值
38         compareAndSetState(c, c + SHARED_UNIT)) {
39         //firstReader是不会放到readHolds里的, 这样，在读锁只有一个的情况下，就避免了查找readHolds。
40         if (r == 0) {    // 是 firstReader，计数不会放入  readHolds。
41             firstReader = current;
42             firstReaderHoldCount = 1;
43         } else if (firstReader == current) {    //firstReader重入
44             firstReaderHoldCount++;
45         } else {
46             // 非 firstReader 读锁重入计数更新
47             HoldCounter rh = cachedHoldCounter;    //读锁重入计数缓存，基于ThreadLocal实现
48             if (rh == null || rh.tid != current.getId())
49                 cachedHoldCounter = rh = readHolds.get();
50             else if (rh.count == 0)
51                 readHolds.set(rh);
52             rh.count++;
53         }
54         return 1;
55     }
56     //第一次获取读锁失败，有两种情况：
57     //1）没有写锁被占用时，尝试通过一次CAS去获取锁时，更新失败（说明有其他读锁在申请）
58     //2）当前线程占有写锁，并且有其他写锁在当前线程的下一个节点等待获取写锁，除非当前线程的下一个节点被取消，否则fullTryAcquireShared也获取不到读锁
59     return fullTryAcquireShared(current);
60 }

 1 //FairSync中需要判断是否有前驱节点，如果有则返回false，否则返回true。遵循FIFO
 2 final boolean readerShouldBlock() {
 3     return hasQueuedPredecessors();
 4 }
 5 final boolean readerShouldBlock() {
 6     return apparentlyFirstQueuedIsExclusive();
 7 }
 8 //当head节点不为null且head节点的下一个节点s不为null且s是独占模式（写线程）且s的线程不为null时，返回true。
 9 //目的是不应该让写锁始终等待。作为一个启发式方法用于避免可能的写线程饥饿，这只是一种概率性的作用，因为如果有一个等待的写线程在其他尚未从队列中出队的读线程后面等待，那么新的读线程将不会被阻塞。
10 final boolean apparentlyFirstQueuedIsExclusive() {
11     Node h, s;
12     return (h = head) != null &&
13         (s = h.next)  != null &&
14         !s.isShared()         &&
15         s.thread != null;
16 }

 1 final int fullTryAcquireShared(Thread current) {
 2     /*
 3      * This code is in part redundant with that in
 4      * tryAcquireShared but is simpler overall by not
 5      * complicating tryAcquireShared with interactions between
 6      * retries and lazily reading hold counts.
 7      */
 8     HoldCounter rh = null;
 9     for (;;) {
10         int c = getState();
11         //如果当前线程不是写锁的持有者，直接返回-1，结束尝试获取读锁，需要排队去申请读锁
12         if (exclusiveCount(c) != 0) {
13             if (getExclusiveOwnerThread() != current)
14                 return -1;
15             // else we hold the exclusive lock; blocking here
16             // would cause deadlock.
17         //如果需要阻塞，说明除了当前线程持有写锁外，还有其他线程已经排队在申请写锁，故，即使申请读锁的线程已经持有写锁（写锁内部再次申请读锁，俗称锁降级）还是会失败，因为有其他线程也在申请写锁，此时，只能结束本次申请读锁的请求，转而去排队，否则，将造成死锁。
18         } else if (readerShouldBlock()) {
19             // Make sure we're not acquiring read lock reentrantly
20             if (firstReader == current) {
21                 //如果当前线程是第一个获取了写锁，那其他线程无法申请写锁
22                 // assert firstReaderHoldCount > 0;
23             } else {
24                 //从readHolds中移除当前线程的持有数，然后返回-1，然后去排队获取读锁。
25                 if (rh == null) {
26                     rh = cachedHoldCounter;
27                     if (rh == null || rh.tid != current.getId()) {
28                         rh = readHolds.get();
29                         if (rh.count == 0)
30                             readHolds.remove();
31                     }
32                 }
33                 if (rh.count == 0)
34                     return -1;
35             }
36         }
37         if (sharedCount(c) == MAX_COUNT)
38             throw new Error("Maximum lock count exceeded");
39         if (compareAndSetState(c, c + SHARED_UNIT)) {
40             //示成功获取读锁，后续就是更新readHolds等内部变量，
41             if (sharedCount(c) == 0) {
42                 firstReader = current;
43                 firstReaderHoldCount = 1;
44             } else if (firstReader == current) {
45                 firstReaderHoldCount++;
46             } else {
47                 if (rh == null)
48                     rh = cachedHoldCounter;
49                 if (rh == null || rh.tid != current.getId())
50                     rh = readHolds.get();
51                 else if (rh.count == 0)
52                     readHolds.set(rh);
53                 rh.count++;
54                 cachedHoldCounter = rh; // cache for release
55             }
56             return 1;
57         }
58     }
59 }

 1 public  void unlock() {
 2     sync.releaseShared(1);
 3 }
 4 
 5 public final boolean releaseShared(int arg) {
 6     if (tryReleaseShared(arg)) {
 7         doReleaseShared();
 8         return true;
 9     }
10     return false;
11 }
12 
13 protected final boolean tryReleaseShared(int unused) {
14     Thread current = Thread.currentThread();
15     //更新计数
16     if (firstReader == current) {
17         // assert firstReaderHoldCount > 0;
18         if (firstReaderHoldCount == 1)
19             firstReader = null;
20         else
21             firstReaderHoldCount--;
22     } else {
23         HoldCounter rh = cachedHoldCounter;
24         if (rh == null || rh.tid != current.getId())
25             rh = readHolds.get();
26         int count = rh.count;
27         if (count <= 1) {
28             readHolds.remove();
29             if (count <= 0)
30                 throw unmatchedUnlockException();
31         }
32         --rh.count;
33     }
34     //自旋CAS，减去1<<16
35     for (;;) {
36         int c = getState();
37         int nextc = c - SHARED_UNIT;
38         if (compareAndSetState(c, nextc))
39             // Releasing the read lock has no effect on readers,
40             // but it may allow waiting writers to proceed if
41             // both read and write locks are now free.
42             return nextc == 0;
43     }
44 }