概述
ReentrantReadWriteLock是Lock的另一種實現方式烈和，我們已經知道了ReentrantLock是一個排他鎖，同一時間只允許一個線程訪問皿淋，而ReentrantReadWriteLock允許多個讀線程同時訪問招刹，但不允許寫線程和讀線程恬试、寫線程和寫線程同時訪問。相對于排他鎖疯暑，提高了并發(fā)性训柴。在實際應用中，大部分情況下對共享數據（如緩存）的訪問都是讀操作遠多于寫操作妇拯，這時ReentrantReadWriteLock能夠提供比排他鎖更好的并發(fā)性和吞吐量幻馁。
另外
1.ReentrantReadWriteLock支持鎖的降級，即先獲取寫鎖越锈，再獲取讀鎖仗嗦，再釋放寫鎖。
2.讀鎖不支持Condition甘凭，會拋出UnsupportedOperationException異常稀拐，寫鎖支持Condition。
讀鎖的獲取
state的高16位讀鎖總共獲取的次數（包括每個線程重入的次數）丹弱，對于每個讀線程的重入次數保存在ThreadLocalHoldCounter中钩蚊。
低16位保存寫鎖的狀態(tài)。
protected final int tryAcquireShared(int unused) {
/*
* Walkthrough:
* 1. If write lock held by another thread, fail.
* 2. Otherwise, this thread is eligible for
* lock wrt state, so ask if it should block
* because of queue policy. If not, try
* to grant by CASing state and updating count.
* Note that step does not check for reentrant
* acquires, which is postponed to full version
* to avoid having to check hold count in
* the more typical non-reentrant case.
* 3. If step 2 fails either because thread
* apparently not eligible or CAS fails or count
* saturated, chain to version with full retry loop.
*/
Thread current = Thread.currentThread();
int c = getState();
//體現鎖降級的思想蹈矮，如果寫鎖被占用砰逻，并且占用寫鎖的線程不是當前線程，返回泛鸟。
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
//保存第一個獲取到讀鎖的線程
firstReader = current;
//保存第一個獲取讀鎖的線程的重入的次數
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
//保存最近獲取讀鎖的線程
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
}
return 1;
}
return fullTryAcquireShared(current);
}
讀鎖是否需要阻塞蝠咆，在公平鎖中，如果同步隊列中有阻塞的節(jié)點就阻塞北滥，在非公平鎖中刚操，如果隊列中有寫線程節(jié)點就阻塞，目的是防止寫線程饑餓再芋。
final boolean readerShouldBlock() {
/* As a heuristic to avoid indefinite writer starvation,
* block if the thread that momentarily appears to be head
* of queue, if one exists, is a waiting writer. This is
* only a probabilistic effect since a new reader will not
* block if there is a waiting writer behind other enabled
* readers that have not yet drained from the queue.
*/
return apparentlyFirstQueuedIsExclusive();
}
final boolean apparentlyFirstQueuedIsExclusive() {
Node h, s;
return (h = head) != null &&
(s = h.next) != null &&
!s.isShared() &&
s.thread != null;
}
如果讀線程需要阻塞菊霜，或者獲取資源失敗，執(zhí)行fullTryAcquireShared
final int fullTryAcquireShared(Thread current) {
/*
* This code is in part redundant with that in
* tryAcquireShared but is simpler overall by not
* complicating tryAcquireShared with interactions between
* retries and lazily reading hold counts.
*/
HoldCounter rh = null;
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)
return -1;
// else we hold the exclusive lock; blocking here
// would cause deadlock.
} else if (readerShouldBlock()) {
// Make sure we're not acquiring read lock reentrantly
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
if (rh == null) {
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current)) {
rh = readHolds.get();
if (rh.count == 0)
readHolds.remove();
}
}
if (rh.count == 0)
return -1;
}
}
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
if (compareAndSetState(c, c + SHARED_UNIT)) {
if (sharedCount(c) == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
if (rh == null)
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
cachedHoldCounter = rh; // cache for release
}
return 1;
}
}
}
注意tryAcquireShared返回值济赎，返回值大于0表示獲取到資源鉴逞，小于0沒有獲取到資源
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
自旋獲取資源，獨占鎖在獲取到資源以后不會有向下傳遞的行為，共享鎖在獲取到資源以后，會向下傳遞喚醒阻塞的其他共享線程写半。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
//如果頭節(jié)點有變化也就是有其他線程獲取到了資源爱榕，繼續(xù)循環(huán)向下傳遞
if (h == head) // loop if head changed
break;
}
}
對于共享鎖的傳播，如果隊列中有一個非共享節(jié)點，則到此停止傳播。為什么s==null也會執(zhí)行doReleaseShared贫橙？這樣可能會喚醒一些沒必要喚醒的節(jié)點喘帚，但是考慮在這個時間段會有比較多的讀線程畅姊，所以也會執(zhí)行doReleaseShared，對在執(zhí)行doReleaseShared期間加入到當前節(jié)點后面的線程做一次unpark吹由，就可以是后繼節(jié)點不阻塞涡匀，直接獲取資源。
然后將頭節(jié)點狀態(tài)設置為PROPAGATE溉知，保證能夠進入if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0)
讀鎖的釋放
public void unlock() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
} else {
//更新HoldCounter陨瘩，0就移除，不為0減少重入的次數
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
int count = rh.count;
if (count <= 1) {
readHolds.remove();
if (count <= 0)
throw unmatchedUnlockException();
}
--rh.count;
}
//自旋更新state的狀態(tài)
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
// Releasing the read lock has no effect on readers,
// but it may allow waiting writers to proceed if
// both read and write locks are now free.
return nextc == 0;
}
}
