标签：oca vat 控制台 mat split 状态 表示 read 回车
读写锁. 读锁之间是共享的. 写锁是独占的.
首先声明一点: 我在分析源码的时候, 把jdk源码复制出来进行中文的注释, 有时还进行编译调试什么的, 为了避免和jdk原生的类混淆, 我在类前面加了"My". 比如把ReentrantLock改名为了MyReentrantLock, 在源码分析的章节里, 我基本不会对源码进行修改, 所以请忽视这个"My"即可.
unsafe在这里是用来给TID_OFFSET赋值的.
那么TID_OFFSET是什么? 就是tid变量在Thread类里的偏移量. tid就是线程id.
下面就是获取TID_OFFSET的源码: (这里我进行了一点改动, 改为了反射)
同步器:
读锁:
写锁:
这是一个带参构造器, 可以选择公平锁还是非公平锁. 同时实例化了读锁和写锁.
而默认构造器是直接调用上面的带参构造器, 采用了非公平锁:
现在模拟一个场景. 两个线程, 同时申请读锁, 场景的demo如下:
public class Main {
static final Scanner scanner = new Scanner(System.in);
private static ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
private static ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
private static ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
static volatile String cmd = "";
public static void main(String[] args) {
new Thread(Main::funcA).start();
new Thread(Main::funcB).start();
while (scanner.hasNext()) {
cmd = scanner.nextLine();
}
}
public static void funcA() {
blockUntilEquals(() -> cmd, "lock a");
readLock.lock();
System.out.println("funcA获取了读锁");
blockUntilEquals(() -> cmd, "unlock a");
readLock.unlock();
System.out.println("funcA释放了读锁");
}
public static void funcB() {
blockUntilEquals(() -> cmd, "lock b");
readLock.lock();
System.out.println("funcB获取了读锁");
blockUntilEquals(() -> cmd, "unlock b");
readLock.unlock();
System.out.println("funcB释放了读锁");
}
private static void blockUntilEquals(Supplier<String> cmdSupplier, final String expect) {
while (!cmdSupplier.get().equals(expect))
quietSleep(1000);
}
private static void quietSleep(int mills) {
try {
Thread.sleep(mills);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
运行上面这段代码.
然后输入"lock a", 然后按下回车, (不带引号), 线程a就获取到了读锁.
然后输入"lock b", 然后按下回车, (不带引号), 线程b就获取到了读锁.
如下图所示, 蓝字为我输入的内容.
可见, 两个读锁之间不是互斥的, 是可以共享同一个锁的.
接下来咱们让这两个线程a和b 分别释放掉读锁.
输入"unlock a", 然后按下回车, (不带括号) , 然后输入"unlock b", 然后按下回车. 就分别释放了两个锁了. 如下图所示:
我带着大家一起调试. 请在funcA()方法里的readLock.lock()这里打下断点, 然后用debug模式运行.
然后再控制台输入 "lock a" , 注意不带引号, 然后按下回车:
然后就发现代码执行到readLock.lock()处就阻塞了:
按下F7 , 进入readLock.lock()方法, 可以看到读锁的lock方法的实现:
可以看到, 调用了acquireShared方法来以共享模式申请了锁.
acquireShared方法源代码如下:
咱们按F7(Step Into进入tryAcquireShared方法, 看看里面的执行过程吧:
protected final int tryAcquireShared(int unused) {// 参数没用
// 获取当前线程的引用
Thread current = Thread.currentThread();
// 获取锁的状态. c 的低 16 位值，代表写锁的状态. 高16位代表读锁的状态
int c = getState();
// exclusiveCount(c) 是写锁的state. 不等于 0，说明有线程持有写锁. (当前场景下肯定等于0, 所以跳过这段if语句)
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// 读锁的state
int r = sharedCount(c);
// 读锁获取是否应该被阻塞, 其实就是根据`等待队列`来判断是否应该被阻塞的 ( 当前场景下没有比当前线程等待更久的线程, 所以不会被阻塞.)
if (!readerShouldBlock() &&
// 判断是否会溢出 (2^16-1). (当前的r==0, 所以没有溢出)
r < MAX_COUNT &&
// 下面这行 CAS 是将 state 属性的高 16 位加 1，低 16 位不变，如果成功就代表获取到了读锁. (当前场景下, 没有线程竞争, 所以肯定成功.)
compareAndSetState(c, c + SHARED_UNIT)) {
/* ----------------------
* 进到这里就是获取到了读锁
* ----------------------*/
// r == 0 说明此线程是第一个获取读锁的，或者说在它之前来的读锁的都走光了. (当前场景r就是等于0, 所以会执行这段if)
if (r == 0) {
// 记录 firstReader 为当前线程.
firstReader = current;
// 持有的读锁数量为1
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, 表示获取到了1个锁.
return 1;
}
return fullTryAcquireShared(current);
}
然后点击这个按钮`放行`:
就发现控制台输出了 "funcA获取了读锁" :
咱们在funcB函数的这句话上也打个断点:
然后再控制台输入 "lock b", 然后按下回车:
按下回车后, 就发现代码阻塞在了刚才的断点上面(红色行变为了绿色):
然后咱们开始分析线程b是如果获取读锁的(记住刚才a线程的读锁还没释放呢), 按下F7, 进入到lock()的源代码:
再F7,
再F7, 终于到了关键的地方:
protected final int tryAcquireShared(int unused) {// 参数没用
// 获取当前线程的引用
Thread current = Thread.currentThread();
// 获取锁的状态. c 的低 16 位值，代表写锁的状态. 高16位代表读锁的状态
int c = getState();
// exclusiveCount(c) 是写锁的state. 不等于 0，说明有线程持有写锁 (当前场景下肯定等于0, 所以跳过这段if语句)
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// 读锁的state, 由于刚才a线程获取到了读锁, 所以这个计数器现在的值是1
int r = sharedCount(c);
// 读锁获取是否应该被阻塞, 其实就是根据`等待队列`来判断是否应该被阻塞的 ( 当前场景下没有比当前线程等待更久的线程, 所以不会被阻塞.)
if (!readerShouldBlock() &&
// 判断是否会溢出 (2^16-1). (当前的r==1, 所以没有溢出)
r < MAX_COUNT &&
// 下面这行 CAS 是将 state 属性的高 16 位加 1，低 16 位不变，如果成功就代表获取到了读锁 (当前场景下, 没有线程竞争, 所以肯定成功.)
compareAndSetState(c, c + SHARED_UNIT)) {
/* ----------------------
* 进到这里就是获取到了读锁
* ----------------------*/
// 当前场景下 r == 1, 而且也不是读锁重入. 所以执行else语句
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
// cachedHoldCounter 用于缓存最后一个获取读锁的线程 (当前场景下, cachedHoldCounter并没有被赋值过, 所以是null)
HoldCounter rh = cachedHoldCounter;
// 当前场景下cachedHoldCounter为空, 所以进入到这个if语句中.
if (rh == null ||
rh.tid != getThreadId(current))
// 利用threadlocal进行创建, 并返回给cachedHoldCounter 和 rh
cachedHoldCounter = rh = readHolds.get();
// 本场景下不执行这个else if, 跳过.
else if (rh.count == 0)
readHolds.set(rh);
// 本场景下, rh刚刚被初始化, 里面的count肯定是0, 在这里进行自增操作, 之后就变为了1.
rh.count++;
}
// return 1, 表示本次tryAcquireShared获取到了1个锁
return 1;
}
return fullTryAcquireShared(current);
}
咱们来总结一下这一段代码都干什么了吧. 首先通过cas操作, 将读锁的state计数器加了1, (也就是变为了2). 然后就是通过ThreadLocal.get() 方法, 在threadlocal里创建了一个b线程的计数器, 并且把这个计数器置为1. 然后就没了....(代码看起来很多的样子, 但是实际上没干多少事情...)
然后将断点放行.(从此以后就不详细讲调试过程了. 就只用语言表述了.)
回到咱们的例子Main方法.在funcA函数的unlock()那一行打上断点. 然后再控制台输入 "unlock a", 然后回车:
然后咱们就可以开始分析 线程a 释放读锁的过程了, 按F7进入到unlock()函数内部:
再F7 :
咱们先看看tryReleaseShared方法吧:
protected final boolean tryReleaseShared(int unused) {// 参数没用
// 获取当前线程的引用
Thread current = Thread.currentThread();
// 判断当前线程是不是当前读锁中的第一个读线程, (线程a就是第一个获取到读锁的, 所以满足这个if条件.)
if (firstReader == current) {
assert firstReaderHoldCount > 0;
// 当前场景下等于 1，所以这次解锁后, 当前线程就不会再持有锁了，把 firstReader 置为 null，给后来的线程用
if (firstReaderHoldCount == 1)
// 为什么不顺便设置 firstReaderHoldCount = 0？因为没必要，其他线程使用的时候自己会设值
firstReader = null;
// 不会执行这个else.
else
firstReaderHoldCount--;
// 不会执行这个else语句. 跳过.
} 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();
// state 的高 16 部分位减 1 , 低16位不动. (高16位是共享模式)
// 高16位的部分, 现在是2. 在这一步减去了1, 所以执行完下面这行代码后 nextc == 1
int nextc = c - SHARED_UNIT;
// cas 更新 state的值为nextc, (当前场景下也就是 1 了), 当前场景下没有争抢, cas肯定成功.
if (compareAndSetState(c, nextc))
// 释放读锁, 对读线程们没有什么影响
// 但如果是 nextc == 0，那就是 state 全部 32 位都为 0，也就是读锁和写锁都空了
// 此时这里返回 true 的话，其实是帮助唤醒后继节点中的获取写锁的线程
// 当前场景下, nextc等于1.所以返回false.
return nextc == 0;
}
}
这段代码最终返回了false, 然后回到上一层函数. 由于返回了false, 所以不会进入到if语句里, 也就是不会执行doReleaseShared()方法:
然后点击`放行`按钮. funcA的读锁释放过程就到此结束了.
回到Main方法. 咱们在funcB里的unlock()函数那一行打上断点. 在控制台输入"unlock b", 然后回车.
然后调试, 一直进入到tryReleaseShared()方法. 刚才讲了tryReleaseShared释放线程a持有的读锁的步骤. 咱们现在看看线程b执行这段代码会有什么不同吧:
protected final boolean tryReleaseShared(int unused) {// 参数没用
// 获取当前线程的引用
Thread current = Thread.currentThread();
// 判断当前线程是不是当前读锁中的第一个读线程,(本场景中, 当然不是了, 而且刚才释放a的读锁的时候, firstReader被设置为了null, 所以也不满足if. 就是说不管读锁a之前是否释放了, 这里都不会满足if条件)
if (firstReader == current) {
assert firstReaderHoldCount > 0;
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
// 会执行这个else语句, 而不是上面的if语句.
} else {
HoldCounter rh = cachedHoldCounter;
// 判断cachedHoldCounter是不是空, 当前场景下cachedHoldCounter不是空, 所以跳过这个if语句.
if (rh == null
|| rh.tid != getThreadId(current))
rh = readHolds.get();
// 获取cachedHoldCounter的计数器, 当前是 1
int count = rh.count;
// 如果计数器小于等于1, 说明该释放了.(目前满足这个if条件, 所以会执行if代码块)
if (count <= 1) {
// 这一步将 ThreadLocal中当前线程对应的计数器 remove 掉，防止内存泄漏。因为已经不再持有读锁了
readHolds.remove();
// 没锁还要释放? 给你抛个异常...
if (count <= 0) throw unmatchedUnlockException();
}
// 计数器 减 1
--rh.count;
}
for (; ; ) {
int c = getState();
// state 的高 16 部分位减 1 , 低16位不动. (高16位是共享模式), 执行完下面这行的减1操作后, nextc就变为0了.
int nextc = c - SHARED_UNIT;
// cas 设置 state
if (compareAndSetState(c, nextc))
// 释放读锁, 对读线程们没有什么影响
// 但如果是 nextc == 0，那就是 state 全部 32 位都为 0，也就是读锁和写锁都空了
// 此时这里返回 true 的话，其实是帮助唤醒后继节点中的获取写锁的线程
// 目前nextc是0, 所以会返回true.
return nextc == 0;
}
}
最终本段代码返回了true, 回到上层代码, 由于返回了true, 所以会执行if代码块里的doReleaseShared()方法:
接下来, 咱们看看doReleaseShared()方法都做了什么事情吧:
由于没有线程进入过`等待队列`, 所以等待队列的head还是null, 所以直接就break了, 什么都没干.
本小节的demo, 就到此结束了.
场景如下: 线程a获取读锁 -> 线程b获取读锁 -> 线程a获取写锁 -> 线程a释放写锁 -> 线程a释放读锁 -> 线程b释放读锁.
1. 场景demo: (还是那句话, 想运行我程序的, 把MyReentrantReadWriteLock 改为JDK的 ReentrantReadWriteLock 就好了. My*系列的都是我复制的JDK代码, 然后改了个名字而已)
import java.util.Scanner;
import java.util.function.Supplier;
public class Main {
static final Scanner scanner = new Scanner(System.in);
static volatile String cmd = "";
private static MyReentrantReadWriteLock lock = new MyReentrantReadWriteLock(true);
private static MyReentrantReadWriteLock.ReadLock readLock = lock.readLock();
private static MyReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
public static void main(String[] args) {
new Thread(Main::funcA).start();
new Thread(Main::funcA2).start();
new Thread(Main::funcB).start();
while (scanner.hasNext()) {
cmd = scanner.nextLine();
}
}
public static void funcA() {
blockUntilEquals(() -> cmd, "lock read a");
readLock.lock();
System.out.println("funcA获取了读锁");
blockUntilEquals(() -> cmd, "unlock read a");
readLock.unlock();
System.out.println("funcA释放了读锁");
}
public static void funcA2(){
blockUntilEquals(() -> cmd, "lock write a");
writeLock.lock();
System.out.println("funcA获取了写锁");
blockUntilEquals(() -> cmd, "unlock write a");
writeLock.unlock();
System.out.println("funcA释放了写锁");
}
public static void funcB() {
blockUntilEquals(() -> cmd, "lock read b");
readLock.lock();
System.out.println("funcB获取了读锁");
blockUntilEquals(() -> cmd, "unlock read b");
readLock.unlock();
System.out.println("funcB释放了读锁");
}
private static void blockUntilEquals(Supplier<String> cmdSupplier, final String expect) {
while (!cmdSupplier.get().equals(expect))
quietSleep(1000);
}
private static void quietSleep(int mills) {
try {
Thread.sleep(mills);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
首先是a获取读锁, 接下来是b获取读锁. 这个场景在上小节中将读锁的时候已经讲过了. 所以这里一代而过.
运行上面这个场景demo, 然后按下进行输入, 来让a线程获取读锁, 然后让b线程获取读锁:
把断点打在writeLock.lock()方法上, 然后输入"lock write a", 按下回车, 来让a线程获取写锁:
发现线程阻塞在了writeLock.lock()方法上. 咱们开始一遍调试一遍分析代码.
F7, 进入到了ReentrantReadWriteLock.WriteLock类里的lock()方法:
接下来就很熟悉了, 根ReentrantLock里的申请锁是同一段代码:
但还不是完全一样, 因为ReadWriteLock重写了其中的tryAcquire方法:
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
// 获取写锁的重入次数, w 在本场景中等于0
int w = exclusiveCount(c);
// c==0说明, 写锁和读锁都没有.
if (c != 0) {
// c != 0 && w == 0: 写锁可用，但是有线程持有读锁(也可能是自己持有) , 在本场景中, 会满足w==0的条件, 而进入if语句
if (w == 0 ||
current != getExclusiveOwnerThread())
// 返回true
return false;
// ***********************************************************
// ---- 本场景根下面的代码没关系, 因为会在上一行的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;
}
然后就是执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) . 这句代码的内部实现与ReentrantLock的代码一模一样(就是同一段代码). 就不再复述了.
执行完这句话之后, 刚刚的申请写锁的线程就被挂起了, 等待着读锁释放完了后唤醒他.
咱们知道他是通过调用AQS类里的parkAndCheckInterrupt方法来进行挂起操作的. 咱们在挂起操作的下一行打个断点. 这样, 到时候这个线程被唤醒后, 咱们就可以感知到了:
接下来咱们在funcB函数里的unlock()方法上打个断点:
接下来咱们释放掉读锁a, 然后释放掉读锁b, 然后线程就会在funcB函数里的unlock()方法上阻塞. (释放这两个锁的流程在前文中已经讲过了, 所以下面简单描述):
按F7, 进入函数内部, 一步一步调试, 最终会执行到 doReleaseShared()方法:
这里其实就是在唤醒`等待队列`里的第一个写锁.
在这里点击`放行`. (点击`放行`就是:"让剩余的函数自动执行完, 一直执行到下一个断点")
就会发现跳转到这里了: (也就是刚才进入等待队列的那个申请写锁的线程从挂起状态恢复到了运行状态)
咱们在此点击`放行`按钮, 然后这个写锁就申请完了:
接下来咱们看看写锁的释放过程. 在funcA2函数里的unlock()方法上打上断线, 然后再控制台输入"unlock write a", 并回车:
然后老规矩按F7, 进入函数内部.查看源码:
再进入一层:
首先是尝试释放锁, 如果锁可以完全释放的话, 就会激活`等待队列`里的第一个线程.
咱们看看读写锁的tryRelease方法的内部实现吧:
其实就是计数器减1, 然后如果等于0的话, 就返回true.表示锁释放干净了. 没有重入.
然后回到上层方法, 由于返回的是true, 所以会进入到if语句里, 然后去判断是否还有线程要获取锁. 如果有的话就用unparkSuccessor方法唤醒. 如果没有的话就直接返回true, 然后结束:
咱们在这一小节 分析一下读锁进入等待队列的流程, 和读锁在等待队列中被唤醒的流程.
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;
import java.util.concurrent.locks.Lock;
import java.util.function.Supplier;
public class Main {
static final Scanner scanner = new Scanner(System.in);
static volatile String cmd = "";
private static MyReentrantReadWriteLock lock = new MyReentrantReadWriteLock(true);
private static MyReentrantReadWriteLock.ReadLock readLock = lock.readLock();
private static MyReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
public static void main(String[] args) {
for (Map.Entry<String, Lock> entry : new HashMap<String, Lock>() {{
put("r1", readLock);
put("r2", readLock);
put("r3", readLock);
put("w1", writeLock);
put("w2", writeLock);
put("w3", writeLock);
}}.entrySet()) {
new Thread(() -> func(entry::getValue, entry.getKey())).start();
}
// 下面这四行, 等价于上面的for循环.
// new Thread(() -> func(() -> readLock, "r1")).start();
// new Thread(() -> func(() -> readLock, "r2")).start();
// new Thread(() -> func(() -> writeLock, "w1")).start();
// new Thread(() -> func(() -> writeLock, "w2")).start();
while (scanner.hasNext()) {
cmd = scanner.nextLine();
}
}
public static void func(Supplier<Lock> myLockSupplier, String name) {
String en_type = myLockSupplier.get().getClass().getSimpleName().toLowerCase().split("lock")[0];
String zn_type = (en_type.equals("read") ? "读" : "写");
blockUntilEquals(() -> cmd, "lock " + en_type + " " + name);
myLockSupplier.get().lock();
System.out.println(name + "获取了" + zn_type + "锁");
blockUntilEquals(() -> cmd, "unlock " + en_type + " " + name);
myLockSupplier.get().unlock();
System.out.println(name + "释放了" + zn_type + "锁");
}
private static void blockUntilEquals(Supplier<String> cmdSupplier, final String expect) {
while (!cmdSupplier.get().equals(expect))
quietSleep(1000);
}
private static void quietSleep(int mills) {
try {
Thread.sleep(mills);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
运行这段代码后, 按下面这样进行输入:
首先是有一个线程申请了写锁w2, 然后是有两个线程分别申请了读锁 r1 和 r2. 等到w2被释放的时候, r1 r2 都申请到了锁.
(输入的时候, 不要打错字, 很容易打错的.)
重新运行这段程序.先输入"lock write w1" , 先申请写锁. 然后在func方法内部的`myLockSupplier.get().lock();`这一行代码打上断点. 然后输入"lock read r1", 申请读锁.
(由于先申请了写锁, 而且这个写锁还没有释放. 所以这个时候申请读锁就意味着会进入`等待队列`)
然后按下F7, 进入到源代码中: (ReentrantReadWriteLock的内部类ReadLock类里的lock()方法)
继续按F7, 进入到方法内部, 咱们就看到了acquireShared方法:
由于刚才咱们成功申请了写锁, 而且还没释放. 所以这次读锁肯定申请失败.
也就是说tryAcquireShared方法尝试获取锁会失败. 失败了就会返回-1. tryAcquireShared方法之前讲过了, 就不细讲了.
tryAcquireShared失败了, 就会满足if条件. 然后就会进入到if语句中执行doAcquireShared方法. 咱们继续往下分析这个doAcquireShared :
private void doAcquireShared(int arg) {
// 将节点放入`等待队列`
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (; ; ) {
final Node p = node.predecessor();
// 判断当前是不是等待队列中的第一个
if (p == head) {
// 尝试获取锁
int r = tryAcquireShared(arg);
if (r >= 0) {
// 把自己设置为新的头部, 然后看看是否需要向后续蔓延
// (也就是, 如果是Shared模式, 那么就会把后续连续的读锁线程都唤醒)
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
其中主要的是setHeadAndPropagate方法. 咱们进入查看源码:
在这里将`等待队列`里的第一个节点设置为了Head节点. 然后判断是不是下一个节点是不是共享模式的, 也就是判断下一个节点是不是共享模式, 如果是的话, 就会执行doReleaseShared()方法. 最终会导致, 一个读锁获取成功的时候, 会带着其后续连续的读锁都一起获取成功.
其中的doReleaseShared()方法在前面小节已经介绍过了, 就不讲了. 如果哪里遗漏了就后续补充.
[源码分析]读写锁ReentrantReadWriteLock
标签：oca vat 控制台 mat split 状态 表示 read 回车
原文地址：https://www.cnblogs.com/noKing/p/9371238.html
