联博开奖_JDK源码剖析-ReentrantReadWriteLock
发表时间：2021-01-05 浏览量：92
原创
JDK源码剖析-ReentrantReadWriteLock
概述
前面剖析过 ReentrantLock「JDK源码剖析-ReentrantLock」，它是一种互斥的可重入锁，可用于处置并发场景下的线程安全问题。而许多时刻会泛起“读多写少”的情形，若用 ReentrantLock 会降低并发量，此时就对照适合 ReentrantReadWriteLock 进场了。
ReentrantReadWriteLock 是读写锁，它维护了一对锁：一个读锁，一个写锁。读锁之间是共享的，写锁是互斥的。与 ReentrantLock 相比，读写锁在读多写少的场景下允许更高的并发量。它的类署名如下：
public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {}下面剖析其代码实现。
代码剖析
ReadWriteLock 接口
ReentrantReadWriteLock 实现了 ReadWriteLock 接口，其代码如下：
public interface ReadWriteLock {
/**
* 返回读锁
*/
Lock readLock();
/**
* 返回写锁
*/
Lock writeLock();
}该接口界说了两个方式，划分返回读锁和写锁，有关 Lock 接口的剖析可参考前文「JDK源码剖析-Lock&Condition」。
组织器
仍然先从组织器最先剖析，如下：
// 无参组织器（默认非公正）
public ReentrantReadWriteLock() {
this(false);
}
// 以给定的公正计谋建立一个 ReentrantReadWriteLock 工具
// true 为公正，false 为非公正
public ReentrantReadWriteLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}与 ReentrantLock 类似，这里的组织器也传入了公正计谋，且默以为非公正。组织器内部初始化了三个变量：sync、readerLock 和 writerLock，如下：
// 提供读锁的内部类
private final ReentrantReadWriteLock.ReadLock readerLock;
// 提供写锁的内部类
private final ReentrantReadWriteLock.WriteLock writerLock;
// 执行所有的同步机制
final Sync sync;下面先剖析这几个内部类的代码。
Sync 类
Sync 类继续自 AQS（与 ReentrantLock 中的 Sync 类似），如下：
abstract static class Sync extends AbstractQueuedSynchronizer {
// 使用 AQS 中的 state 变量（int 类型）来纪录读写锁的占用情形
// 其中高 16 位纪录读锁的持有次数；低 16 位纪录写锁的重入次数
static final int SHARED_SHIFT = 16;
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
// 共享锁（读锁）的持有次数
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
// 互斥锁（写锁）的重入次数
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
/**
* 每个线程持有读锁的计数器。
* 以 ThreadLocal 形式保留，缓存在 cachedHoldCounter
* 该类的主要作用是纪录线程持有读锁的数目，可明白为 <tid,count> 的形式
*/
static final class HoldCounter {
int count = 0;
// Use id, not reference, to avoid garbage retention
final long tid = getThreadId(Thread.currentThread());
}
static final class ThreadLocalHoldCounter
extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
/**
* 当前线程持有的可重入读锁的数目（数目为0时删除）。
*/
private transient ThreadLocalHoldCounter readHolds;
private transient HoldCounter cachedHoldCounter;
/**
* firstReader：第一个获取读锁的线程；
* firstReaderHoldCount：firstReader 的持有计数。
*/
private transient Thread firstReader = null;
private transient int firstReaderHoldCount;
Sync() {
readHolds = new ThreadLocalHoldCounter();
setState(getState()); // ensures visibility of readHolds
}
/*
* 对于公正锁和非公正锁，获取和释放锁使用的代码相同；
* 但在行列非空时，它们是否或若何允许插入的方式差别。
*/
/**
* 当前线程在实验（或有资格）获取读锁时，是否应该由于计谋原因而壅闭。
*/
abstract boolean readerShouldBlock();
/**
* 当前线程在实验（或有资格）获取写锁时，是否应该由于计谋原因而壅闭。
*/
abstract boolean writerShouldBlock();
// 释放写锁
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;
}
// 获取写锁
protected final boolean tryAcquire(int acquires) {
/*
* 流程：
* 1. 若其他线程持有读锁或写锁（计数不为零），返回 false；
* 2. 若持有数目饱和（超出上限），返回 false；
* 3. 该线程有资格获取锁，更新 state 并设置为 owner。
*/
Thread current = Thread.currentThread();
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;
}
// 若获取写锁时应该壅闭，或者更新 state 失败，返回 false
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}
// 释放读锁
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;
}
// 更新 state
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;
}
}
// 获取读锁
protected final int tryAcquireShared(int unused) {
/*
* 流程：
* 1. 若是其他线程持有写锁，获取失败；
* 2. 否则，该线程有资格获取，因此请询问它是否由于行列计谋而壅闭；
* 若不壅闭，实验通过 CAS 更新状态计数。
* 注重：这一步没有检查可重入的获取，推迟到完整版本，
* 以制止在显著不能重入的情形下检查持有计数。
* 3. 若是第二步失败，要么是因为线程显著不符合条件、CAS 失败或计数饱和，
* 则举行完整重试版本。
*/
Thread current = Thread.currentThread();
int c = getState();
// step1. 若写锁被其他线程占用，则获取失败
// exclusiveCount(c) != 0示意写锁被占用
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// step2. 获取读锁数目
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
// 读锁未被占用，设置该线程是第一个持有读锁的线程
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
// 该线程已持有读锁，计数加1
} else if (firstReader == current) {
firstReaderHoldCount++;
// 其他线程已持有读锁
} else {
// 取缓存
HoldCounter rh = cachedHoldCounter;
// 若未初始化，或者拿到的不是当前线程的计数，则从 ThreadLocal 中获取
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
// 增添计数
rh.count++;
}
// 获取乐成
return 1;
}
// step3. 若step2获取失败，则执行该步骤
return fullTryAcquireShared(current);
}
/**
* 获取读锁的完整版，处置 tryAcquireShared 中未处置的 CAS 丢失和可重入读取。
*/
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;
}
}
}
/**
* 执行写锁的 tryLock 方式
* 与 tryAcquire 相比，该方式未挪用 writerShouldBlock
*/
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;
}
/**
* 执行读锁的 tryLock 方式
* 与 tryAcquireShared 相比，该方式未挪用 readerShouldBlock
*/
final boolean tryReadLock() {
Thread current = Thread.currentThread();
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return false;
int r = sharedCount(c);
if (r == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
if (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 true;
}
}
}
}Sync 类继续自 AQS，主要重写了 AQS 中独占模式（参考「JDK源码剖析-AbstractQueuedSynchronizer(2)」）和共享模式（参考「JDK源码剖析-AbstractQueuedSynchronizer(3)」）下获取和释放锁的方式。
Sync 类的继续结构如下：
NonfairSync 类
NonfairSync 继续自 Sync 类，提供非公正计谋的实现，如下：
static final class NonfairSync extends Sync {
final boolean writerShouldBlock() {
return false; // writers can always barge
}
final boolean readerShouldBlock() {
// 挪用父类 AQS 中的方式实现
return apparentlyFirstQueuedIsExclusive();
}
}
// 若头节点的下一个节点是写线程，为了防止写线程饥饿守候，当前的读线程应该壅闭
final boolean apparentlyFirstQueuedIsExclusive() {
Node h, s;
return (h = head) != null &&
(s = h.next) != null &&
!s.isShared() &&
s.thread != null;
}非公正计谋中，writerShouldBlock 返回 false，说明写线程无需壅闭；
readerShouldBlock 则是挪用父类 AQS 中的 apparentlyFirstQueuedIsExclusive 方式实现的，该方式通过判断守候行列中的第一个线程是否为写线程，若是则返回 true，示意给写线程让道。
PS: 通过剖析这两个方式，发现在非公正计谋下，写线程的优先级照样高于读线程的（纯属小我私家明白）。
,,www.9cx.net采用以太坊区块链高度哈希值作为统计数据，联博以太坊统计数据开源、公平、无任何作弊可能性。联博统计免费提供API接口，支持多语言接入。
FairSync 类
FairSync 也继续自 Sync 类，提供公正计谋的实现，如下：
static final class FairSync extends Sync {
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
}在公正计谋中，两个方式都通过挪用父类 AQS 的 hasQueuedPredecessors 方式判别，二者都是凭据守候行列中是否有其他线程，若有其他线程，则当前线程守候。
这就是公正的体现吧：无论读写，都乖乖去排队，别插队。
ReadLock
ReadLock 是读锁的实现，代码如下：
public static class ReadLock implements Lock, java.io.Serializable {
private final Sync sync;
protected ReadLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
/**
* 获取读锁（不能中止）：
* 1. 若其他线程未持有写锁，则获取读锁并立刻返回；
* 2. 若其他线程持有写锁，则由于线程调剂，当前线程被禁用并休眠，直到获取读锁。
*/
public void lock() {
sync.acquireShared(1);
}
/**
* 以中止方式获取读锁：
* 1. 若其他线程未持有写锁，则获取读锁并立刻返回；
* 2. 若其他线程持有写锁，则由于线程调剂，当前线程被禁用并休眠，
* 直到当前获取到读锁，或者被其他线程中止。
*/
public void lockInterruptibly() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* 仅当另一个线程未持有写锁时才气获取读锁。
* 若另一个线程持有写锁，则立刻返回 false。
*/
public boolean tryLock() {
return sync.tryReadLock();
}
/**
* 获取读锁（与 tryLock 方式类似，多了超时守候）。
*/
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
/**
* 实验释放该锁。
* 若读线程的数目为零，则该锁可用于实验获取写锁。
*/
public void unlock() {
sync.releaseShared(1);
}
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}ReadLock 类实现了 Lock 接口，它的主要方式就是 Lock 接口所界说的方式（获取和释放锁）。读锁之间是共享的，ReadLock 的主要方式都通过 AQS 共享模式的方式实现的。
WriteLock
WriteLock 是写锁，代码如下：
public static class WriteLock implements Lock, java.io.Serializable {
private final Sync sync;
protected WriteLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
/**
* 获取写锁。
* 1. 若无其他线程持有读锁或写锁，则获取写锁并立刻放回，并将写锁计数设为1；
* 2. 若当前线程已经持有写锁，则将其计数加1，并立刻返回（可重入）；
* 3. 若锁被其他线程持有，当前线程被禁用并处于休眠状态，直到获取写锁（计数设为1）。
*/
public void lock() {
sync.acquire(1);
}
/**
* 获取写锁（可被中止）。
* 1. 若无其他线程持有读锁或写锁，则获取并立刻返回写锁，并将计数设为1；
* 2. 若当前线程已经持有写锁，则将其计数加1，并立刻返回（可重入）；
* 3. 若锁被其他线程持有，当前线程被禁用并处于休眠状态，
* 直到当前线程获取写锁（计数设为1）或被其他线程中止。
*/
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
/**
* 仅当挪用时其他线程未持有该写锁时，才获取该写锁。
*/
public boolean tryLock( ) {
return sync.tryWriteLock();
}
/**
* 实验获取写锁（响应中止，有超时守候）。
*/
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
/**
* 实验释放锁。
* 若当前线程是锁的持有者，则持有计数将削减；若当前持有计数为零则释放锁。
* 若当前线程不是锁的持有者，则抛出异常IllegalMonitorStateException
*/
public void unlock() {
sync.release(1);
}
public Condition newCondition() {
return sync.newCondition();
}
/**
* 查询此写锁是否由当前线程持有。
*/
public boolean isHeldByCurrentThread() {
return sync.isHeldExclusively();
}
/**
* 查询当前线程对该写锁的持有计数。
*/
public int getHoldCount() {
return sync.getWriteHoldCount();
}
}与 ReadLock 类似，WriteLock 类也实现了 Lock 接口，其主要方式也是 Lock 接口所界说的方式（获取和释放锁）。而写锁是互斥的，WriteLock 的大部分方式都是通过 AQS 独占模式的方式实现的。
ReentrantReadWriteLock 的主要代码就剖析到这里，下面简朴剖析其用法和使用场景。
典型用法
示例代码
为便于明白读写锁的操作，下面举个栗子验证（代码仅供参考）：
public class TestRDLock {
// 建立一个线程池
private static ExecutorService threadPoolExecutor = new ThreadPoolExecutor(10, 20,
60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
// 建立一个读写锁实例
private static final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
// 这里可以实验“读读“、“读写”和“写写”场景的代码测试（仅供参考）
threadPoolExecutor.execute(new ReadTask());
threadPoolExecutor.execute(new WriteTask());
}
}
// 写操作
private static class WriteTask implements Runnable {
@Override
public void run() {
readWriteLock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName() + " 获取写锁");
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
readWriteLock.writeLock().unlock();
System.out.println(Thread.currentThread().getName() + " 释放了写锁");
}
}
}
// 读操作
private static class ReadTask implements Runnable {
@Override
public void run() {
readWriteLock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName() + " 获取读锁");
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
readWriteLock.readLock().unlock();
System.out.println(Thread.currentThread().getName() + " 释放了读锁");
}
}
}
}Java API 文档中还提供了两个典型的使用场景，如下：
场景一：更新缓存后执行锁降级
class CachedData {
Object data;
volatile boolean cacheValid;
final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
void processCachedData() {
// 先获取读锁
rwl.readLock().lock();
if (!cacheValid) {
// Must release read lock before acquiring write lock
rwl.readLock().unlock();
rwl.writeLock().lock();
try {
// Recheck state because another thread might have
// acquired write lock and changed state before we did.
// 更新缓存（持有写锁的情形下）
if (!cacheValid) {
data = ...
cacheValid = true;
}
// Downgrade by acquiring read lock before releasing write lock
rwl.readLock().lock();
} finally {
// 释放写锁（仍然持有读锁，即降级为读锁）
rwl.writeLock().unlock(); // Unlock write, still hold read
}
}
try {
use(data);
} finally {
// 释放读锁
rwl.readLock().unlock();
}
}
}场景二：在较大的聚集中，读多写少的情形
class RWDictionary {
private final Map<String, Data> m = new TreeMap<String, Data>();
private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
private final Lock r = rwl.readLock();
private final Lock w = rwl.writeLock();
public Data get(String key) {
r.lock();
try { return m.get(key); }
finally { r.unlock(); }
}
public String[] allKeys() {
r.lock();
try { return m.keySet().toArray(); }
finally { r.unlock(); }
}
public Data put(String key, Data value) {
w.lock();
try { return m.put(key, value); }
finally { w.unlock(); }
}
public void clear() {
w.lock();
try { m.clear(); }
finally { w.unlock(); }
}
}小结
- ReentrantReadWriteLock 是一种读写锁，它持有一对锁：读锁和写锁。其中读锁之间是共享的，写锁是互斥的。
- 「读多写少」的场景下，ReentrantReadWriteLock 比 ReentrantLock 有更高的并发性。
- 与 ReentrantLock 原理类似，ReentrantReadWriteLock 内部也基于 AQS：其中读锁基于「共享模式」实现，写锁基于「独占模式」实现。
参考：
- https://docs.oracle.com/javase/8/docs/api/index.html
- https://blog.csdn.net/fxkcsdn/article/details/82217760
©著作权归作者所有：来自51CTO博客作者WriteOnRead的原创作品，如需转载，请注明出处，否则将追究法律责任
java
0
珍藏
上一篇：JDK源码剖析-CopyOnWr...