前面分析过 ReentrantLock「JDK源码分析-ReentrantLock」,它是一种互斥的可重入锁,可用于处理并发场景下的线程安全问题。而很多时候会出现“读多写少”的情况,若用 ReentrantLock 会降低并发量,此时就比较适合 ReentrantReadWriteLock 出场了。ReentrantReadWriteLock 是读写锁,它维护了一对锁:一个读锁,一个写锁。读锁之间是共享的,写锁是互斥的。与 ReentrantLock 相比,读写锁在读多写少的场景下允许更高的并发量。它的类签名如下:public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {}
ReentrantReadWriteLock 实现了 ReadWriteLock 接口,其代码如下:public interface ReadWriteLock {
Lock readLock();
Lock writeLock();
}
public ReentrantReadWriteLock() {
this(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 类继承自 AQS(与 ReentrantLock 中的 Sync 类似),如下:abstract static class Sync extends AbstractQueuedSynchronizer {
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; }
static final class HoldCounter {
int count = 0;
final long tid = getThreadId(Thread.currentThread());
}
static final class ThreadLocalHoldCounter
extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
private transient ThreadLocalHoldCounter readHolds;
private transient HoldCounter cachedHoldCounter;
private transient Thread firstReader = null;
private transient int firstReaderHoldCount;
Sync() {
readHolds = new ThreadLocalHoldCounter();
setState(getState());
}
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) {
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
if (c != 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;
}
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) {
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))
return nextc == 0;
}
}
protected final int tryAcquireShared(int unused) {
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);
}
final int fullTryAcquireShared(Thread current) {
HoldCounter rh = null;
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)
return -1;
} else if (readerShouldBlock()) {
if (firstReader == current) {
} 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;
}
return 1;
}
}
}
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;
}
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;
}
}
}
}
NonfairSync 继承自 Sync 类,提供非公平策略的实现,如下:static final class NonfairSync extends Sync {
final boolean writerShouldBlock() {
return false;
}
final boolean readerShouldBlock() {
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: 通过分析这两个方法,发现在非公平策略下,写线程的优先级还是高于读线程的(纯属个人理解)。FairSync 也继承自 Sync 类,提供公平策略的实现,如下:static final class FairSync extends Sync {
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
}
在公平策略中,两个方法都通过调用父类 AQS 的 hasQueuedPredecessors 方法判别,二者都是根据等待队列中是否有其他线程,若有其他线程,则当前线程等待。这就是公平的体现吧:无论读写,都乖乖去排队,别插队。public static class ReadLock implements Lock, java.io.Serializable {
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);
}
public boolean tryLock() {
return sync.tryReadLock();
}
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 共享模式的方法实现的。public static class WriteLock implements Lock, java.io.Serializable {
private final Sync sync;
protected WriteLock(ReentrantReadWriteLock lock) {
sync = lock.sync;
}
public void lock() {
sync.acquire(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));
}
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) {
rwl.readLock().unlock();
rwl.writeLock().lock();
try {
if (!cacheValid) {
data = ...
cacheValid = true;
}
rwl.readLock().lock();
} finally {
rwl.writeLock().unlock();
}
}
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(); }
}
}
1. ReentrantReadWriteLock 是一种读写锁,它持有一对锁:读锁和写锁。其中读锁之间是共享的,写锁是互斥的。2. 「读多写少」的场景下,ReentrantReadWriteLock 比 ReentrantLock 有更高的并发性。3. 与 ReentrantLock 原理类似,ReentrantReadWriteLock 内部也基于 AQS:其中读锁基于「共享模式」实现,写锁基于「独占模式」实现。1. https://docs.oracle.com/javase/8/docs/api/index.html2. https://blog.csdn.net/fxkcsdn/article/details/82217760