【源码解析】聊聊ReentrantReadWriteLock是如何实现的读写锁
为什么需要读写锁
在并发编程领域，有多线程进行提升整体性能，但是却引入了共享数据安全性问题。基本就是无锁编程下的单线程操作，有互斥同步锁操作，但是性能不高，并且同一时刻只有一个线程可以操作资源类。但是对于大多数常见下，都是读操作多，写操作少，那么可以利用将锁的粒度进行细化，进而分化出读锁/写锁。也就是syn/ReentrantLock的升级版本ReentrantReadWriteLock。
之前一篇文章已经简单介绍过 ，本篇主要从源码角度剖析具体原理如何实现的。
聊聊ReentrantReadWriteLock锁降级和StampedLock邮戳锁
源码解析
带着三个问题去梳理
- 读写锁是怎样实现分别记录读写锁的状态？
- 读锁如何获取和释放锁
- 写锁如何获取和释放锁
可以看到顶层通过接口定义规范，内部持有Sync实现AQS，分别实现不同的公平锁和非公平锁。
//读写锁的接口规范
public interface ReadWriteLock {
Lock readLock();
Lock writeLock();
}
// 内部持有读写锁
public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {
private static final long serialVersionUID = -6992448646407690164L;
private final ReentrantReadWriteLock.ReadLock readerLock;
private final ReentrantReadWriteLock.WriteLock writerLock;
final Sync sync;
public ReentrantReadWriteLock() {
this(false);
}
默认是非公平锁。内部通过构造方法创建两个锁，读锁和写锁。
public ReentrantReadWriteLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
锁状态
看到这里其实有点懵逼，什么 这都是什么操作，其实在AQS内部通过一个变量state进行控制是否可以获取资源，但是读写锁如何要用两个变量的话，其实不太好，所以就通过高16位代表读锁的状态、低16位代表写锁的状态。
对于低16来说，值等于0没有加写锁，值等于1 加了写锁，大于1 标识写锁的重入次数。
高16来说，0 ：没有加读锁， 1: 加读锁。 值大于1 不表示读锁的重入次数，表示读锁总共被获取了多少次。读锁的重入次数存储在和线程相关的地方，通过threadLocal进行存储。
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 6317671515068378041L;
// 偏移位数
static final int SHARED_SHIFT = 16;
// 共享锁基本单位 左移16位 state+= shared_unit
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
// 读锁、写锁 可重入最大数量
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
// 获取低16位的条件
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
/** 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 */
// 写锁 是否持有 1 为一个线程持有 2 1次冲入 1次获取写锁
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
写状态，等于 S & 0x0000FFFF(将高 16 位全部抹去)。 当写状态加1，等于S+1.
读状态，等于 S >>> 16 (无符号补 0 右移 16 位)。当读状态加1，等于 S+(1<<16),也就是S+0x00010000。
这样 我们就完成了一个state值可以同时表示两种状态的。
写锁
写锁加锁
public void lock() {
sync.acquire(1);
}
调用AQS的获取
public final void acquire(int arg) {
//tryAcquire(arg) true 获取锁成功直接结束
//如果没有获取到锁，acquireQueued 会将线程压入队列中
//!tryAcquire(arg) 没有获取到锁，将当前线程挂起
//addWaiter
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
ReentrantReadWriteLock内部实现了tryAcquire方法。
该方法主要的作用就是
1.获取当前线程
2.判断state的状态。 c = 0 说明当前没有读锁和写锁，通过CAS进行设置state的值 直接获取锁
3.state值不等于0，w == 0 说明当前有读锁 获取锁失败，返回
4.w != 0 说明 当前是写锁重入，所以判断是否最大值，设置state的值+1
writerShouldBlock() 方法会根据是否是公平锁进行排队处理
protected final boolean tryAcquire(int acquires) {
// 获取当前线程
Thread current = Thread.currentThread();
// 获取state的值
int c = getState();
int w = exclusiveCount(c);
// c = 0 说明 当前没有读锁和写锁
if (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");
// Reentrant acquire
// 设置当前值 说明可重入
setState(c + acquires);
return true;
}
// 是否需要阻塞 公平锁
if (writerShouldBlock() ||
//CAS 设置c的值 c += 1
!compareAndSetState(c, c + acquires))
return false;
// 设置为当前线程
setExclusiveOwnerThread(current);
return true;
}
写锁释放锁
当当前线程执行完毕业务逻辑之后，就会释放锁。
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
//唤醒阻塞等待的线程
unparkSuccessor(h);
return true;
}
return false;
}
释放锁的流程主要就是
1.判断持有锁的线程是否属于当前线程，不是直接异常
2.将state-1 ,state = 0的话，说明重入的锁释放完毕。清空
3.设置state的值，可能是-1 或者 为0。
protected final boolean tryRelease(int releases) {
// 持有锁的线程 是否等于当前线程
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 将当前state -= 1
int nextc = getState() - releases;
boolean free = exclusiveCount(nextc) == 0;
// 如果写锁为0 说明当前没有锁持有了
if (free)
// 将当前线程释放
setExclusiveOwnerThread(null);
// 设置state的值
setState(nextc);
return free;
}
读锁
读锁加锁
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
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 &&
//CAS 设置 高16位加1
compareAndSetState(c, c + SHARED_UNIT)) {
// 第一次获取读锁
if (r == 0) {
//设置第一个获取读锁的线程
firstReader = current; // 当前线程
//设置第一个获取读锁线程的重入数
firstReaderHoldCount = 1; //
} else if (firstReader == current) {
// 如果当前线程是第一个获取读锁的线程，重入数++
firstReaderHoldCount++;
} else {
//刷新除获取锁的第一个读线程的重入数
// threadLocal进行记录线程重入次数
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);
}
从这里可以看到，支持锁降级，持有写锁的线程，可以获取读锁，但是后续要记得把读锁和写锁读释放
读锁释放锁
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) {
// 第一个获取读锁的线程 重入次数等于=1
// assert firstReaderHoldCount > 0;
if (firstReaderHoldCount == 1)
//第一个获取读锁的线程设置为null
firstReader = null;
else
// 当前线程重入多次 -1
firstReaderHoldCount--;
//如果不是第一个获取读锁的线程，获取该线程的锁重入次数对象
} else {
// 获取线程持有共享锁的数量对象
HoldCounter rh = cachedHoldCounter;
// 如果rh==null 当前线程不是共享锁数量对象对应的线程id
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;
}
//CAS同步更新
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;
}
}
线程读锁的重入数与读锁数量是两个概念，线程读锁的重入数是每个线程获取同一个读锁的次数，读锁数量则是所有线程的读锁重入数总和。举一个例子就是 3个线程 分别获取了3次读锁，那么读锁数量就是9，每个线程的读锁重入数就是3。
锁升级&锁降级
锁升级就是线程持有读锁的前提下，去升级为写锁，显然这是违背读写互斥的。
锁降级，线程持有写锁的前提下，降级为读锁。
好了我们来看为什么需要锁降级，如果说针对一块临界区直接加一把大锁，那么其实并发读很低，那么可不可以在获取写锁的前提下 降级为读锁，这样既保证数据的一致性，又可以提升整体的并发度。锁降级就是为了结局这个问题。
设计思想
通过本篇的大概学习，我们了解到RRW中几个设计要点，通过一个变量去控制两个读写锁的状态，位运算的方式。值得我们借鉴，另一种就是锁降级的为了保证数据安全。以及在整体的代码实现上大量使用模板模式，AQS的子类都是相同的方式。
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！