1. 结构
uml依赖关系
其核心还是队列同步器Sync，它被ReadLock和WriteLock所共享。state为锁标记， 其中高16位为ReadLock总计数标记，低16位为WriteLock（独占模式）的重入计数。
2. 读锁
2.1.读锁获取
public void lock() {
sync.acquireShared(1);
}
/*
* 1. 若写锁被其他线程持有， 则失败
* 2. 若当前线程可以锁定写状态，先判断是否应当阻塞(排队策略)， 不需阻塞的话，尝试CAS地修改状态位。
* 3. 若2修改失败了，可能是因为其他线程此时已经持有了写锁，也可能是 CAS失败, 还有可能是其他未处理的可重入读(具体可以看代码注释),那么应该走fullyTryAcquireShared流程，不断重试
*/
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
// 1.若写锁被其他线程持有， 则失败
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
int r = sharedCount(c);
// 2.若当前线程可以锁定写状态，先判断是否应当阻塞(排队策略)， 不需阻塞的话，尝试CAS地修改状态位。
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
// 修改成功，若当前线程是第一个读线程(第一次进入或者重入)
if (r == 0) { // 第一次进入
firstReader = current;
firstReaderHoldCount = 1; // 记录重入次数
} else if (firstReader == current) { // 它就是第一个读线程，发生重入， 将其重入次数+1
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;
}
/*
* step2失败，可能原因是：
* CAS尝试失败,
* rh!= null && rh.tid != currentThreadId && rh.count > 0， 这意味着除了firstReader外，还有其他reader将rh状态修改了
*/
return fullTryAcquireShared(current);
}
如果tryAcquireShared()中第2步失败了，那么进入fullTryAcquireShared()不断重试。
final int fullTryAcquireShared(Thread current) {
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;
}
}
}
2.3 读锁释放
public void unlock() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
// 若读锁被彻底释放，那么需要去唤醒队列中排队获取X锁的线程
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
/*
1. 如果当前线程就是firstReader, 那么将其重入计数减1， 减到0后将firstReader置为null
2. 否则，获取当前线程的重入计数， 将其减1， 减到0将记录当前线程重入次数的readHold移除
3. 死循环内，CAS地修改读锁计数(state高16位)
返回： 读锁获取计数是否为0
*/
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 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;
}
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;
}
}
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
}
if (h == head) // loop if head changed
break;
}
}
3. 写锁
3.1 写锁结构
写锁与读锁共享同一个队列同步器， 本质上写锁就是一把X锁。
3.2 写锁获取
public void lock() {
sync.acquire(1);
}
/*
* 先尝试获取，尝试失败，则进入阻塞队列，等待获取
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
// 排队等待获取锁
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
// 补偿中断标记，其原因是在acquireQueued中，每次unpark后都要清除中断标记
selfInterrupt();
}
/*
* 1. 如果当前有其他线程获取了读锁或写锁， 那么本次获取失败
* 2. c ！= 0 && w != 0, 表示当前线程要重入，如果独占锁重入计数过大，那么返回失败；否则将重入计数增加
* 3. 当前线程有资格去修改state， 但具体本次能不能改取决于排队策略： 比如fair mode下，有其他线程在排队获取X锁，那么就不能直接获取
*/
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
// state 高16位为shared标记位， 低16位为exclusive lock 重入计数。
int c = getState();
int w = exclusiveCount(c);
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire
setState(c + acquires);
return true;
}
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}
addWaiter()和acquireQueued()实质上就是AQS的方法，与ReentrantLock一样，在这里不啰嗦了。
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
3.3 写锁释放
public void unlock() {
sync.release(1);
}
release()中unparkSuccessor(Node node)为AQS方法，同ReentrantLock,不做赘述。
public final boolean release(int arg) {
// 如果tryRelease返回true， 则意味着重入计数为0了，那么应该考虑唤醒等待队列中的线程
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
// 减掉重入计数，返回是否完全释放(free)
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
boolean free = exclusiveCount(nextc) == 0;
if (free)
setExclusiveOwnerThread(null);
setState(nextc);
return free;
}
