Jdk1.6 JUC源码解析(8)-locks-ReentrantReadWriteLock
作者：大飞
功能简介：
- ReentrantReadWriteLock提供了读写锁的机制，读锁使用AQS的共享模式，写锁使用独占模式。
- ReentrantReadWriteLock也支持公平/非公平锁。
- ReentrantReadWriteLock的写锁会阻塞读锁和写锁，读锁只会阻塞写锁。
- ReentrantReadWriteLock的写锁可以降级为读锁，但读锁不能升级为写锁。
源码分析：
- ReentrantReadWriteLock实现了ReadWriteLock接口，先来看下这个接口：
public interface ReadWriteLock {
/**
* Returns the lock used for reading.
*
* @return the lock used for reading.
*/
Lock readLock();
/**
* Returns the lock used for writing.
*
* @return the lock used for writing.
*/
Lock writeLock();
}
- ReentrantReadWriteLock内部也是基于AQS构建同步机制，先看下这部分：
/**
* 用于实现ReentrantReadWriteLock的同步机制。
* 下面有公平和非公平版本的子类。
*/
static abstract class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 6317671515068378041L;
/*
* AQS中的state在ReentrantReadWriteLock被分为2部分：
* 高16位bit来表示读锁的持有(重入)次数，低16位表示写锁的持有次数。
*/
static final int SHARED_SHIFT = 16;
static final int SHARED_UNIT = (1 << SHARED_SHIFT); // 00000000 00000001 00000000 00000000
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; // 00000000 00000000 11111111 11111111
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; // 00000000 00000000 11111111 11111111
/** Returns the number of shared holds represented in count */
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count */
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
...
}
之前分析AQS时候，提到了AQS开放了几个方法交由子类去实现，那就从这个几个方法着手看起。首先看下tryAcquire方法实现：
protected final boolean tryAcquire(int acquires) {
/*
* 独占模式请求的过程:
* 1. 如果锁持有者不是当前线程，且读锁或者写锁持有
* 次数不为0，请求失败。
* 2. 如果锁持有次数超过上限，请求失败。
* 3. 否则，如果当前是一个重入请求，或者队列策略允许，
* 那么当前线程可以获取锁，更新状态并设置锁所有者。
*/
Thread current = Thread.currentThread();
int c = getState();
//获取写锁持有次数。
int w = exclusiveCount(c);
if (c != 0) {
//如果c不等于0且w等于0，说明读锁持有次数不等于0。
if (w == 0 || current != getExclusiveOwnerThread())
return false;
//如果写锁持有次数超过最大次数，抛错。
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 否则，重入流程。
setState(c + acquires);
return true;
}
//如果c为0，说明当前没有任何线程持有锁。
//首先判断是否应该阻塞写锁，如果不需要阻塞，CAS来设置state，如果设置成功，继续设置锁持有者，请求成功；
//如果写锁需要被阻塞或CAS操作失败，请求失败。
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}
看一下compareAndSetState这个方法：
/**
* Returns true if the current thread, when trying to acquire
* the write lock, and otherwise eligible to do so, should block
* because of policy for overtaking other waiting threads.
*/
abstract boolean writerShouldBlock();这个方法在本类中并未实现，而是交给子类(公平和非公平版本)去实现。
接下来看下tryRelease方法：
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//写锁重入次数减1
int nextc = getState() - releases;
//如果减1后，写锁重入次数为0，返回true(这样在AQS中就会唤醒同步队列里的线程)；否则返回false。
boolean free = exclusiveCount(nextc) == 0;
if (free)
setExclusiveOwnerThread(null);
setState(nextc);
return free;
}
继续看下tryAcquireShared方法：
protected final int tryAcquireShared(int unused) {
/*
* 共享模式的请求流程:
* 1. 如果有其他线程持有写锁，方法失败。
* 2. 如果重入次数超过最大值，抛错。
* 3. 否则，判断当前队列策略是否需要阻塞读请求，
* 如果不需要，尝试CAS修改状态，修改成功的话，
* 更新重入次数，请求成功；
* 4. 如果第3步失败，调用fullTryAcquireShared方法。
*/
Thread current = Thread.currentThread();
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
//如果写锁重入次数不为0，且锁所有者不是当前线程，请求失败。
return -1;
if (sharedCount(c) == MAX_COUNT) //如果读锁重入次数超过最大值，抛错。
throw new Error("Maximum lock count exceeded");
//如果当前不允许阻塞读锁，那么尝试增加读锁重入次数。
if (!readerShouldBlock() &&
compareAndSetState(c, c + SHARED_UNIT)) {
//获取缓存的HoldCounter。
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != current.getId())
//如果当前没有缓存HoldCounter，或者缓存的不是当前线程的HoldCounter。
//从缓存中获取当前线程的HoldCounter(没有会新建一个)。
cachedHoldCounter = rh = readHolds.get();
//缓存HoldCounter计数增加。
rh.count++;
return 1; //获取成功。
}
//上面的条件路径都失败，那么会继续调用fullTryAcquireShared方法。
return fullTryAcquireShared(current);
}
看下readerShouldBlock方法：
/**
* Returns true if the current thread, when trying to acquire
* the read lock, and otherwise eligible to do so, should block
* because of policy for overtaking other waiting threads.
*/
abstract boolean readerShouldBlock();这个方法在本类中并未实现，而是交给子类(公平和非公平版本)去实现。
看下fullTryAcquireShared方法：
/**
* Full version of acquire for reads, that handles CAS misses
* and reentrant reads not dealt with in tryAcquireShared.
*/
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
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != current.getId())
rh = readHolds.get();
for (;;) {
int c = getState();
int w = exclusiveCount(c);
if ((w != 0 && getExclusiveOwnerThread() != current) ||
((rh.count | w) == 0 && readerShouldBlock()))
//如果有其他线程持有当前写锁，或者没有线程持有当前写锁且当前线程没有持有读锁且读锁需要阻塞，那么请求失败。
return -1;
if (sharedCount(c) == MAX_COUNT) //如果写锁重入次数超出最大值，抛错。
throw new Error("Maximum lock count exceeded");
//尝试获取读锁。
if (compareAndSetState(c, c + SHARED_UNIT)) {
//获取成功后，设置缓存HoldCounter
cachedHoldCounter = rh; // cache for release
//HoldCounter技术累加。
rh.count++;
//读请求成功。
return 1;
}
}
}
再看下tryReleaseShared方法：
protected final boolean tryReleaseShared(int unused) {
//缓存当前线程对应的HoldCounter
HoldCounter rh = cachedHoldCounter;
Thread current = Thread.currentThread();
if (rh == null || rh.tid != current.getId())
rh = readHolds.get();
if (rh.tryDecrement() <= 0) //通过rh检测是否出现没请求就释放的情况。
throw new IllegalMonitorStateException();
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
//读锁重入次数减1，如果读锁释放，返回true。
return nextc == 0;
}
}
上面几个方法里都出现了HoldCounter，来看看HoldCounter类及相关的代码：
/**
* 用于统计每个线程持有读锁次数的计数器。
* 在ThreadLocal中管理; 缓存在cachedHoldCounter字段中。
*/
static final class HoldCounter {
int count;
// 使用id，而不是引用，是为了避免垃圾滞留。
final long tid = Thread.currentThread().getId();
/** Decrement if positive; return previous value */
int tryDecrement() {
int c = count;
if (c > 0)
count = c - 1;
return c;
}
}
/**
* ThreadLocal subclass. Easiest to explicitly define for sake
* of deserialization mechanics.
*/
static final class ThreadLocalHoldCounter
extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
/**
* The number of read locks held by current thread.
* Initialized only in constructor and readObject.
*/
transient ThreadLocalHoldCounter readHolds;
/**
* The hold count of the last thread to successfully acquire
* readLock. This saves ThreadLocal lookup in the common case
* where the next thread to release is the last one to
* acquire. This is non-volatile since it is just used
* as a heuristic, and would be great for threads to cache.
*/
transient HoldCounter cachedHoldCounter;
Sync() {
readHolds = new ThreadLocalHoldCounter();
setState(getState()); // 这里做一次volatile读写，相当于加了一个内存屏障，确保上面readHolds的可见性。
}
最后还有一个需要覆盖的AQS的方法isHeldExclusively：
protected final boolean isHeldExclusively() {
// While we must in general read state before owner,
// we don't need to do so to check if current thread is owner
return getExclusiveOwnerThread() == Thread.currentThread();
}实现方式和ReentrantLock中一样。
另外，同步机制里还提供了tryWriteLock和tryReadLock方法，来支持非阻塞的tryLock方法。
/**
* 非阻塞写锁的请求方法, 两种模式下都可以插队。
* 除了没有调用writerShouldBlock之外，和tryAcquire实现一致。
*/
final boolean tryWriteLock() {
Thread current = Thread.currentThread();
int c = getState();
if (c != 0) {
int w = exclusiveCount(c);
if (w == 0 ||current != getExclusiveOwnerThread())
return false;
if (w == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
}
if (!compareAndSetState(c, c + 1))
return false;
setExclusiveOwnerThread(current);
return true;
}
/**
* 非阻塞读锁的请求方法, 两种模式下都可以插队。
* 除了没有调用writerShouldBlock之外，和tryAcquireShared 实现一致。
*/
final boolean tryReadLock() {
Thread current = Thread.currentThread();
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return false;
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
if (compareAndSetState(c, c + SHARED_UNIT)) {
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != current.getId())
cachedHoldCounter = rh = readHolds.get();
rh.count++;
return true;
}
}
}
还有另外一些给ReentrantReadWriteLock实现提供支持的方法。
final ConditionObject newCondition() {
return new ConditionObject();
}
final Thread getOwner() {
// Must read state before owner to ensure memory consistency
return ((exclusiveCount(getState()) == 0)?
null :
getExclusiveOwnerThread());
}
final int getReadLockCount() {
return sharedCount(getState());
}
final boolean isWriteLocked() {
return exclusiveCount(getState()) != 0;
}
final int getWriteHoldCount() {
return isHeldExclusively() ? exclusiveCount(getState()) : 0;
}
final int getReadHoldCount() {
return getReadLockCount() == 0? 0 : readHolds.get().count;
}
接下来看下公平版的子类实现：
/**
* Fair version of Sync
*/
final static class FairSync extends Sync {
private static final long serialVersionUID = -2274990926593161451L;
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
}可见，结合上面同步机制基类看，这里的判断读写请求是否应该阻塞的实现就是检查AQS同步等待队列中有没有比当前线程还早的线程。
再看下非公平版的子类实现：
/**
* Nonfair version of Sync
*/
final static class NonfairSync extends Sync {
private static final long serialVersionUID = -8159625535654395037L;
final boolean writerShouldBlock() {
return false; // 写请求总是可以插队。
}
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(); //如果AQS同步等待队列头有写请求，那么当前读请求阻塞。
}
}
- 接下来看下内部读写锁的实现。
先看下读锁的实现：
public static class ReadLock implements Lock, java.io.Serializable {
private static final long serialVersionUID = -5992448646407690164L;
private final Sync sync;
protected ReadLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
/**
* 请求读锁。如果没有其他线程持有写锁，那么请求成功，方法返回。
* 如果有其他线程持有写锁，那么当前线程阻塞，直到请求成功(被唤醒后尝试请求)。
*/
public void lock() {
sync.acquireShared(1);
}
/**
* 如上面的过程相似，只是如果线程被中断，抛出异常。
*/
public void lockInterruptibly() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* 如果没有其他线程持有写锁，那么请求读锁成功。
*
* 没有其他写锁的情况下，请求会立即成功，就算在公平模式下，
* 也会立即请求成功，不管同步队列里有没有其他线程在等待读锁。
* 也就是说会打破公平规则，但在特定情况下是有利的。
* 如果想遵循公平规则，这样调用：tryLock(0, TimeUnit.SECONDS)。
*/
public boolean tryLock() {
return sync.tryReadLock();
}
/**
* 支持超时，中断的读锁请求。
*
* 没有其他写锁的情况下，请求会立即成功。
* 如果锁被设置为公平策略并且同步队列中有其他线程等待，那么
* 就算当前没有其他线程持有锁，锁也不能被获取。
*
* 如果想使用支持超时的且能插队的公平锁，
* 这样调用：if (lock.tryLock() || lock.tryLock(timeout, unit) ) { ... }
*/
public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
/**
* 释放读锁。
*
* 如果释放读锁后，读锁持有次数为0，那么写锁就可以尝试获取这个锁了。
*/
public void unlock() {
sync.releaseShared(1);
}
/**
* 读锁(共享模式)不支持条件，所以抛出异常。
*
* @throws UnsupportedOperationException always
*/
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}
再看下写锁的实现：
public static class WriteLock implements Lock, java.io.Serializable {
private static final long serialVersionUID = -4992448646407690164L;
private final Sync sync;
protected WriteLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
/**
* 请求写锁。
*
* 如果没有其他线程持有当前锁的读锁或者写锁，那么请求写锁成功，
* 并将写锁持有次数设置为1。
*
* 如果当前线程已经持有了写锁，然后请求成功并增加持有次数。
*
* 如果有其他线程持有当前锁的读锁或者写锁，当前线程阻塞，
* 直到请求锁成功(被其他线程唤醒后继续尝试)。
*/
public void lock() {
sync.acquire(1);
}
/**
* 和上面过程一致，支持中断。
*/
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
/**
* 如果没有其他线程持有当前锁的读锁或者写锁，那么请求读锁成功，返回true。
*
* 会打破公平策略，如果想遵循公平策略，这样调用：tryLock(0, TimeUnit.SECONDS)
*
* 如果当前线程已经持有当前写锁，增加写锁持有次数，返回true。
*
* 如果有其他线程持有当前锁的读锁或者写锁，返回失败。
*/
public boolean tryLock( ) {
return sync.tryWriteLock();
}
/**
* 支持超时和中断。
*
* 没有其他读锁和写锁的情况下，请求会立即成功。
* 如果锁被设置为公平策略并且同步队列中有其他线程等待，那么
* 就算当前没有其他线程持有锁，锁也不能被获取。
*
* 如果想使用支持超时的且能插队的公平锁，
* 这样调用：if (lock.tryLock() || lock.tryLock(timeout, unit) ) { ... }
*/
public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
/**
* 释放写锁。
*
* 如果释放后，当前线程对写锁的持有次数为0，那么其他线程就可以请求这个锁了。
*/
public void unlock() {
sync.release(1);
}
/**
* 返回新建的条件对象。
*/
public Condition newCondition() {
return sync.newCondition();
}
/**
* 判断写锁是否被当前线程持有。
*/
public boolean isHeldByCurrentThread() {
return sync.isHeldExclusively();
}
/**
* 判断当前线程对写锁的持有(重入)次数。
*/
public int getHoldCount() {
return sync.getWriteHoldCount();
}
}
最后看一下ReentrantReadWriteLock中提供的一些支持监控的方法：
/**
* 判断锁是否为公平锁。
*/
public final boolean isFair() {
return sync instanceof FairSync;
}
/**
* 返回当前锁的持有线程(不精确)。
*/
protected Thread getOwner() {
return sync.getOwner();
}
/**
* 获取读锁持有次数。
*/
public int getReadLockCount() {
return sync.getReadLockCount();
}
/**
* 判断是否有线程持有当前锁的写锁。
*/
public boolean isWriteLocked() {
return sync.isWriteLocked();
}
/**
* 判断当前写锁是否被当前线程持有。
*/
public boolean isWriteLockedByCurrentThread() {
return sync.isHeldExclusively();
}
/**
* 获取当前线程对当前写锁的持有次数。
*/
public int getWriteHoldCount() {
return sync.getWriteHoldCount();
}
/**
* 获取当前线程对当前读锁的持有次数。
*/
public int getReadHoldCount() {
return sync.getReadHoldCount();
}
/**
* 获取同步队列中等待写锁的线程(不精确)。
*/
protected Collection<Thread> getQueuedWriterThreads() {
return sync.getExclusiveQueuedThreads();
}
/**
* 获取同步队列中等待读锁的线程(不精确)。
*/
protected Collection<Thread> getQueuedReaderThreads() {
return sync.getSharedQueuedThreads();
}
/**
* 判断同步队列中是否有线程在等待。
*/
public final boolean hasQueuedThreads() {
return sync.hasQueuedThreads();
}
/**
* 判断当前线程是否在同步队列中。
*/
public final boolean hasQueuedThread(Thread thread) {
return sync.isQueued(thread);
}
/**
* 获取当前线程
*/
public final int getQueueLength() {
return sync.getQueueLength();
}
/**
* 获取同步队列中等待线程(不精确)。
*/
protected Collection<Thread> getQueuedThreads() {
return sync.getQueuedThreads();
}
/**
* 判断给定条件上是否有等待线程。
*/
public boolean hasWaiters(Condition condition) {
if (condition == null)
throw new NullPointerException();
if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
throw new IllegalArgumentException("not owner");
return sync.hasWaiters((AbstractQueuedSynchronizer.ConditionObject)condition);
}
/**
* 获取给定条件上等待线程的数量。
*/
public int getWaitQueueLength(Condition condition) {
if (condition == null)
throw new NullPointerException();
if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
throw new IllegalArgumentException("not owner");
return sync.getWaitQueueLength((AbstractQueuedSynchronizer.ConditionObject)condition);
}
/**
* 获取给定条件上等待线程。
*/
protected Collection<Thread> getWaitingThreads(Condition condition) {
if (condition == null)
throw new NullPointerException();
if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
throw new IllegalArgumentException("not owner");
return sync.getWaitingThreads((AbstractQueuedSynchronizer.ConditionObject)condition);
}
ReentrantReadWriteLock的代码解析完毕！
参见：Jdk1.6 JUC源码解析(6)-locks-AbstractQueuedSynchronizer
