简介
ReentrantReadWriteLock是一个悲观的可重入的读写锁,而StampedLock既支持悲观锁也支持乐观锁但不支持锁的重入, 在ReentrantReadWriteLock下如果多个线程同时获取读锁的时候,获取写锁的线程就会被挂起进行等待,在StampedLock乐观锁下如果有线程加了写锁,其它读线程可以获取共享变量的值。
常量
/** 处理器数量 */
private static final int NCPU = Runtime.getRuntime().availableProcessors();
/** 入队前最大的自旋次数 */
private static final int SPINS = (NCPU > 1) ? 1 << 6 : 0;
/** 第一次入队时自旋次数 */
private static final int HEAD_SPINS = (NCPU > 1) ? 1 << 10 : 0;
/** 第一次入队时最大的自旋次数 */
private static final int MAX_HEAD_SPINS = (NCPU > 1) ? 1 << 16 : 0;
/**
* state中的低7位二进制代表读锁加锁次数
* 第8位二进制代表写锁
* */
private static final int LG_READERS = 7;
//读锁加锁时在原有锁状态上加的值,这样就能获取到读锁的版本号已经读锁的状态
//读锁加锁的时候每次在锁状态上的基础值加1
//写锁加锁的时候每次在锁状态上的基础值加256
private static final long RUNIT = 1L;
//写锁所在的二进制位 1000 0000
//1代表写锁
private static final long WBIT = 1L << LG_READERS;
//读锁所在的二进制位 0111 1111
//1代表读锁
private static final long RBITS = WBIT - 1L;
//读锁的二进制位最大加锁次数,因为7个二进制位表达的数值只有这么多
//并不是说读锁只能加锁这么多,溢出的加锁次数会用readerOverflow变量来存放
private static final long RFULL = RBITS - 1L;
//读锁和写锁所代表的二进制位 1111 1111
private static final long ABITS = RBITS | WBIT;
//对读锁的二进制位进行取反
//获取到的是-128,读锁的二进制位为0以外其余的二进制位都为1
//1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1000 0000
//该数为最大的写锁版本号
private static final long SBITS = ~RBITS;
//版本号的初始值 256
private static final long ORIGIN = WBIT << 1;
//等待加锁
private static final int WAITING = -1;
//取消加锁
private static final int CANCELLED = 1;
//读模式
private static final int RMODE = 0;
//写模式
private static final int WMODE = 1;
/** Wait nodes */
static final class WNode {
//上一个节点
volatile WNode prev;
//下一个节点
volatile WNode next;
//挂载的读节点
volatile WNode cowait;
//节点线程
volatile Thread thread;
//节点状态
volatile int status;
//节点模式 读/写
final int mode;
WNode(int m, WNode p) {
mode = m;
prev = p;
}
}
/** 头节点 */
private transient volatile WNode whead;
/** 尾节点 */
private transient volatile WNode wtail;
/** 锁的状态以及版本号 默认256*/
private transient volatile long state;
/** 溢出的读锁加锁次数 */
private transient int readerOverflow;
复制代码NCPU：程序运行的电脑的处理器数量
SPINS：入队前自旋的次数,根据处理器数量来计算,当前自旋次数耗尽时就会为当前线程创建节点并入队
HEAD_SPINS：线程节点入队时自旋的次数,并不是每个线程节点入队时都会自旋,只有在入队的时候,上一个节点是头节点的时候才会自旋,该值也是根据处理器数量来计算的
MAX_HEAD_SPINS：线程节点入队时最大的自旋次数,当HEAD_SPINS自旋次数耗尽时则会将HEAD_SPINS自旋次数翻倍作为下次的自旋次数,最大自旋次数不能超过MAX_HEAD_SPINS,如果已到达最大的自旋次数的时候则使用最大的自旋次数来自旋
LG_READERS：读锁所占的二进制位个数
RUNIT：读锁加锁时需要在state上加的值
WBIT：写锁所在的二进制位
RBITS：读锁所在的二进制位
RFULL：读锁的二进制位最大能表示的加锁次数(126),并不是说读锁只能加锁这么多次,只是因为二进制位表示的只有这么多,溢出的加锁次数会使用readerOverflow变量来存放
ABITS：读锁和写锁所在的二进制位
SBITS：写锁的最大版本号
ORIGIN：版本号的初始值
WAITING：当前线程节点的状态为WAITING则说明后续有线程节点等待加锁
CANCELLED：当前线程节点的状态为CANCELLED则说明当前线程节点已经取消了加锁
RMODE：表示一个线程节点是读模式的节点,也可以理解为这个线程需要加的是读锁
WMODE：表示一个线程节点是写模式的节点,也可以理解为这个线程需要加的是写锁
WNode：队列等待的线程节点
prev：上一个节点
next：下一个节点
cowait：挂载的读节点
thread：节点中加锁的线程
status：节点的状态
mode：节点的模式
whead：队列中的头节点
wtail：队列中的尾节点
state：锁的状态以及版本号
readerOverflow：读锁溢出的加锁次
构造方法
public StampedLock() {
state = ORIGIN;
}
复制代码创建StampedLock的时候将ORIGIN的值赋给了state,而ORIGIN的值是通过WBIT的值左移1位获得的,而WBIT的值将1左移7位获得的值为128,再将128左移1位得到256,最终state和ORIGIN的值为256,256是state初始的锁状态以及版本号。
那是如何用一个state就能定义锁状态和版本号的呢?
state是用long类型来修饰的,占8个字节,1个字节8bit位,总共占64bit位,前56bit位为锁的版本号,第8位为写锁的标识,最后面的7位为读锁的标识。
readLock（读锁）
public long readLock() {
//s 锁状态
//next 锁版本号
long s = state, next;
return ((whead == wtail && (s & ABITS) < RFULL &&
U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ?
next : acquireRead(false, 0L));
}
复制代码-
whead == wtail：校验头节点是否与尾节点相同,如果相同则说明队列中没有线程节点在等待 -
( s & ABITS ) < RFULL：校验是否有线程加了锁,通过当前锁的状态s与ABITS进行与运算获取到锁所在的二进制位,如果二进制位对应的十进制位的数大于RFULL则说明有线程加了写锁,小于则说明没有线程加锁或者是加了读锁,如果没有加锁或加的是读锁那当前线程就可以通过CAS来获取锁,CAS成功则返回锁的版本号next,如果失败或有线程加了写锁那就调用acquireRead方法自旋或阻塞。 -
例1：
s= 256ABITS= 255RFULL= 126二进制：256 = 1 0000 0000 255 = 1111 1111
与运算：1 0000 0000 & 1111 1111 = 0000 0000
(只有两个相同的进制位上的数都为1时运算后的进制位才为1,256的二进制位有9个,而128的进制位只有8个,那该怎么办呢?此时不足的进制位则补0,最后发现运算后的值为0则说明当前没有线程加锁)
-
例2：
s= 257ABITS= 255RFULL= 126二进制：257 = 1 0000 0001 255 = 1111 1111
与运算：1 0000 0001 & 1111 1111 = 0000 0001 转换为十进制 = 1
(经过与运算之后最终的十进制值为1是大于0小于126的,说明有线程加了读锁)
-
例3：
s= 384ABITS= 255RFULL= 126二进制：384 = 1 1000 0000 255 = 1111 1111
与运算：1 1000 0000 & 1111 1111 = 1000 0000 转换为十进制 = 128
(经过与运算之后最终的十进制值为128是大于126的,说明有线程加了写锁)
-
例4：
s= 420ABITS= 255RFULL= 126二进制：420 = 0001 1010 0100 255 = 1111 1111
与运算：0001 1010 0100 & 1111 1111 = 1010 0100 转换为十进制 = 164
(经过与运算之后最终的十进制值为164是大于126的,看样子是加了写锁,但并不是,可以看到与运算后的二进制位上既占了写锁的标识也占了读锁的标识,此时这个锁状态是个错误的,在实际情况中并不会出现这样的错误的,此处只是演示一下)
acquireRead
private long acquireRead(boolean interruptible, long deadline) {
//node 当前线程所在的节点
//p 当前线程节点的上一个节点/原尾节点
WNode node = null, p;
for (int spins = -1; ; ) {
//头节点
WNode h;
//校验头节点是否与尾节点相同,如果头节点与尾节点相同则说明等待队列中没有线程节点在等待加锁
if ((h = whead) == (p = wtail)) {
//s 锁状态
//m 等于0没有线程加锁, 大于0小于128有线程加了读锁,大于128有线程加了写锁
//ns 加锁的版本号
for (long m, s, ns; ; ) {
//先校验锁状态是否被其它线程获取了锁,大于0则说明被获取了锁
//再校验是否小于读锁的最大加锁次数RFULL,如果小于则说明当前线程可以尝试加读锁
//如果大于就分为两种情况：1.有其它线程加了写锁 2.读锁加锁的次数已到RFULL的最大次数
//如果有其它线程加了写锁,那么m肯定是大于WBIT的,则会执行后面的else语句
//如果读锁的加锁次数已到RFULL的最大次数,那么m肯定是小于WBIT的,因为WBIT的值为128
//而读锁的二进制标识为0111 1111,该二进制为127
//而写锁的二进制标识为1000 0000,该二进制为128
//if语句中的三元表达式中如果为true,肯定是没有线程加锁或者是加的读锁
//如果为false要么已经是加的写锁,要么是读锁的加锁次数已经到达了用二进制来标识读锁加锁的次数的最大值
if ((m = (s = state) & ABITS) < RFULL ?
U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) : (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))
//返回加锁的版本号
return ns;
else if (m >= WBIT) {
//有线程加了写锁
if (spins > 0) {
//自旋
if (LockSupport.nextSecondarySeed() >= 0)
--spins;
}
else {
if (spins == 0) {
//当自旋次数耗尽
//nh 头节点
//np 尾节点
WNode nh = whead, np = wtail;
//(头节点没有变更并且尾节点也没有变更)或者(头节点不等于尾节点)则退出循环
if ((nh == h && np == p) || (h = nh) != (p = np))
break;
}
//根据cpu数量计算自旋的次数
spins = SPINS;
}
}
}
}
if (p == null) {
//p为空则说明等待队列中没有节点此时就需要初始化队列节点
//创建一个写模式的节点
WNode hd = new WNode(WMODE, null);
//将创建的节点设置为头和尾节点
if (U.compareAndSwapObject(this, WHEAD, null, hd))
wtail = hd;
}
else if (node == null)
//为当前线程创建一个读模式的节点
//并将节点的上一个节点的指针指向原尾节点
node = new WNode(RMODE, p);
//校验队列中的节点是否变更
else if (h == p || p.mode != RMODE) {
//节点没有变更才会进入当前语句
//再次校验一下节点是否变更
if (node.prev != p)
//队列中的节点变更了,修改当前节点的上一个节点的指针并重新执行循环方法再次校验队列中的节点是否变更
node.prev = p;
//队列中的节点没有变更,则将当前线程节点设置为尾节点并将原尾节点的next指针指向当前线程节点
else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
p.next = node;
break;
}
}
//如果队列中的节点变更了则将当前线程节点挂载到新入队的节点上,并将新入队的节点上的被挂载的节点挂载到当前线程节点上
/**
* 当前节点
* ----- ----- -----
* |node| ----> | p | | p |
* ----- ----- -----
* | c1挂载在p节点上 ——————————> |
* ----- -----
* | c1 | |node|
* ----- -----
* |
* -----
* | c1 |
* -----
* 将当前线程节点node挂载到p节点上,并将p节点上挂载的节点挂载到node节点上
*/
else if (!U.compareAndSwapObject(p, WCOWAIT, node.cowait = p.cowait, node))
//挂载失败则清空node节点上挂载的节点
//此时p节点上挂载的节点不变
node.cowait = null;
//节点变更了并且将当前线程节点挂载到了p节点上
else {
for ( ; ; ) {
//c 挂载的节点
//w 挂载的节点的线程
//pp 上一个节点的上一个节点
WNode pp, c; Thread w;
//如果头节点中挂载着节点则将挂载的节点线程唤醒
if ((h = whead) != null && (c = h.cowait) != null &&
U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
(w = c.thread) != null)
U.unpark(w);
//校验当前节点的上一个节点是否为头节点或上一个节点的上一个节点是头节点
if (h == (pp = p.prev) || h == p || pp == null) {
//m 是否加锁
//s 锁状态
//ns 锁的版本号
long m, s, ns;
do {
//校验当前是否有线程在加锁,如果m小于RFULL则说明没有线程在加锁或者加的是读锁
//此时就可以尝试加读锁,如果m大于等于RFULL则说明线程在加写锁或者是加读锁的次数已经溢出
//如果是加读锁的次数溢出则会调用tryIncReaderOverflow方法将溢出的次数记录下来
//如果是写锁的话,m < WBIT该条件是不会成立的
//如果加的是写锁的话,m是大于WBIT的
if ((m = (s = state) & ABITS) < RFULL ?
U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
(m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))
return ns;
} while (m < WBIT);
}
//有线程加了写锁或上一个节点不是头节点或上上个节点不是头节点
//校验节点是否变更
if (whead == h && p.prev == pp) {
long time;
//pp == null 说明当前节点是头节点或者说上一个节点是头节点
//当前加锁的线程节点排在了对头,此时可以重新走外面的大循环
//如果头节点和尾节点相同则说明队列中已经没有在等待加锁的线程节点了,当前线程就可以尝试的去加锁
//如果头节点和尾节点不相同,可能是在当前线程准备重新走外面的大循环的时候来了新的线程节点
//此时当前线程则需要重新入队进行排队获取锁
//h == p 上一个节点是头节点,那当前线程节点从队列中移除,走外面的大循环尝试加锁
//p.status > 0 上一个节点已经取消了加锁,当前线程节点可能会成为队列中的第二个节点,此时可以尝试去加锁
if (pp == null || h == p || p.status > 0) {
node = null;
break;
}
//校验是否设置了超时时间
if (deadline == 0L)
time = 0L;
//如果设置了超时时间则校验是否超时,如果超时则将当前线程节点从队列中移除
else if ((time = deadline - System.nanoTime()) <= 0L)
return cancelWaiter(node, p, false);
//获取当前线程
Thread wt = Thread.currentThread();
//将Thread类中的阻塞对象修改为当前类
U.putObject(wt, PARKBLOCKER, this);
node.thread = wt;
if ((h != pp || (state & ABITS) == WBIT) && whead == h && p.prev == pp)
//挂起线程
U.park(false, time);
//清除节点中的线程引用
node.thread = null;
//将Thread类中的阻塞对象置空
U.putObject(wt, PARKBLOCKER, null);
//线程是否中断
if (interruptible && Thread.interrupted())
//如果线程被中断了,则将线程节点冲队列中移除
return cancelWaiter(node, p, true);
}
}
}
}
for (int spins = -1; ; ) {
//h 头节点
//np 当前线程节点的上一个节点
//pp 上上个节点
WNode h, np, pp;
//ps 上个节点的状态
int ps;
//校验上一个节点是否是头节点
if ((h = whead) == p) {
if (spins < 0)
//更新自旋次数
spins = HEAD_SPINS;
else if (spins < MAX_HEAD_SPINS)
//已到指定的自旋次数时,还未获取到锁并且自旋的次数未到最大自旋次数则将自旋次数翻倍
spins <<= 1;
//根据自旋次数来自旋
for (int k = spins; ; ) {
//m 是否有线程加锁,锁标识
//s 锁状态
//ns 锁版本号
long m, s, ns;
//校验是否有线程加了锁,如果有线程加了锁则判断锁标识m是否小于RFULL
//如果锁标识m小于RFULL则说明没有线程加锁或者线程加的是读锁
//如果m小于RFULL,当前线程可以尝试去加锁
//如果锁标识m大于等于RFULL,分为两种情况：1.有线程加了写锁 2.加读锁的次数已经溢出
//如果溢出则会校验m是否小于WBIT,如果加的是写锁,m肯定是大于WBIT的
//如果是溢出的读锁那就会调用tryIncReaderOverflow方法尝试获取读锁并将溢出的次数使用新的变量来存放
if ((m = (s = state) & ABITS) < RFULL ?
U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
(m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L)) {
//加读锁成功
//c 挂载的节点
WNode c;
//w 挂载的节点中的线程
Thread w;
//将当前线程节点设置为头节点
whead = node;
node.prev = null;
//获取当前线程节点中挂载的节点并校验节点是否为空
while ((c = node.cowait) != null) {
//不为空则说明当前线程节点中挂载着线程节点
//依次将当前线程节点中挂载的线程节点唤醒
if (U.compareAndSwapObject(node, WCOWAIT, c, c.cowait) && (w = c.thread) != null)
U.unpark(w);
}
//返回锁的版本号
return ns;
}
//加读锁失败或有线程加了写锁,当前线程自旋次数-1
//如果当前线程自旋次数小于等于0则退出自旋
else if (m >= WBIT && LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
break;
}
}
//校验头节点是否不为空
else if (h != null) {
//挂载的节点
WNode c;
//挂载的节点中的线程
Thread w;
//如果头节点中挂载的线程节点不为空则依次将挂载的线程节点唤醒
while ((c = h.cowait) != null) {
if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) && (w = c.thread) != null)
U.unpark(w);
}
}
//校验头节点是否改变
if (whead == h) {
//原尾节点与当前线程节点的上一个节点不相同
//说明之前或后面来了新的线程并创建了新的节点将原尾节点修改了
//此时就需要将原尾节点修改成当前线程节点的上一个节点
//当前线程修改了原尾节点,等后面线程执行的时候会发现原尾节点与线程节点的上一个节点不相同也会进行修改
if ((np = node.prev) != p) {
if (np != null)
(p = np).next = node;
}
else if ((ps = p.status) == 0)
//如果上一个节点的状态为0则将上一个节点的状态修改为-1
//节点状态为-1则说明后续有节点在等待加锁,后续p节点释放了锁之后需要唤醒后续节点
U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
else if (ps == CANCELLED) {
//如果上一个节点的状态为1则说明线程节点已经取消加锁
//则需要将上一个节点从队列中移除
if ((pp = p.prev) != null) {
node.prev = pp;
pp.next = node;
}
}
else {
//尝试加锁失败
long time;
//校验是否设置了加锁的超时时间
if (deadline == 0L)
time = 0L;
//如果设置了超时时间则校验是否超时
else if ((time = deadline - System.nanoTime()) <= 0L)
//超时则将当前线程节点从队列中移除
return cancelWaiter(node, node, false);
//获取当前线程
Thread wt = Thread.currentThread();
//将Thread类中的阻塞对象修改为当前类
U.putObject(wt, PARKBLOCKER, this);
node.thread = wt;
if (p.status < 0 && (p != h || (state & ABITS) == WBIT) && whead == h && node.prev == p)
//挂起线程
U.park(false, time);
//清除节点中的线程引用
node.thread = null;
//将Thread类中的阻塞对象置空
U.putObject(wt, PARKBLOCKER, null);
//线程是否中断
if (interruptible && Thread.interrupted())
//如果线程中断了则将当前线程节点从队列中清除
return cancelWaiter(node, node, true);
}
}
}
}
复制代码acquireRead方法代码比较多,主要分为两个大的for循环,for循环中有着许多的if判断语句,我们一个个的来看,先从第一个for循环开始。
1.(h = whead) == (p = wtail)：头节点是否与尾节点相同,如果相同则说明队列中没有线程节点在等待加锁则执行当前if语句中的代码,不相同则跳过当前语句执行序号2的if语句
1.1.for(long m, s, ns; ; )：队列中没有线程节点,当前线程自旋。
1.1.1. (m=(s=state)&ABITS)<RFULL?CAS : (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L)：s & ABITS < RFULL上面已经讲过了,这里就不再讲了,校验是否有其它线程加锁,如果小于RFULL则说明没有其它线程加锁或其它线程加的是读锁,其它线程加了读锁的话,当前线程也可以去加读锁,则会执行CAS操作尝试获取读锁,执行成功直接返回读锁的版本号,如果大于等于RFULL则校验m是否小于WBIT(写锁标识),如果不小于则说明有线程加了写锁,此时就会执行序号1.1.2的if语句,如果小于则说明当前用二进制来表示读锁的加锁次数已经到达最大值,此时就需要调用tryIncReaderOverflow方法将溢出的加读锁的次数使用readerOverflow变量来存放,如果在执行CAS和tryIncReaderOverflow方法时失败了则会继续从1.1开始执行。
1.1.2.m >= WBIT：校验是否有线程加了写锁,如果线程加了写锁则会校验自旋次数是否大于0,线程第一次执行当前代码时自旋次数肯定是不大于0的,那就会根据cpu的数量来计算自旋的次数,下次再执行校验的时候自旋次数就大于0了,当大于0时就自旋尝试获取锁,直到成功获取锁或自旋次数耗尽则会退出当前循环。
2.p == null：当加锁失败或队列中等待的线程节点不为空才会走到当前校验,如果队列中的线程节点不为空时,此时p也不会为null,p不为null就不会执行当前if语句中的代码,只有当前队列中没有线程节点在等待并且当前线程尝试加锁失败的时候才会执行,当前语句会创建一个写模式的节点并将该节点设置为头节点和尾节点。
3.node == null：当前线程在序号1中自旋获取锁失败时并且当前线程节点为空则会为当前线程创建一个读模式的节点并将上一个节点的指针指向原尾节点。
4.h == p || p.mode != RMODE：h == p说明当前线程没有创建节点之前队列中并没有线程节点在等待,此时就可以将为当前线程创建的节点设置为尾节点并退出第一个大循环执行第二个大循环,h != p说明之前队列中是有线程节点在等待的,此时就需要校验之前尾节点的模式,如果之前尾节点的模式不是读模式,那将当前线程节点设置为尾节点并退出第一个大循环执行第二个大循环,在将当前线程节点设置为尾节点的时候会先校验一下当前线程节点的上一个节点是否变更了,如果变更了就将当前线程节点的上一个节点的指针指向变更的节点继续执行第一个大循环。
5.!U.compareAndSwapObject(p, WCOWAIT, node.cowait = p.cowait, node)：原先队列中就有线程节点在等待并且队列中的尾节点是读模式的节点,此时就需要将当前读模式的线程节点挂载到尾节点上,这样读模式的线程节点能在同一时间被唤醒并获取锁。
6.只有当队列中的节点变更了并且将当前线程节点挂载到了尾节点上才会走到最终的else语句中。
6.1.(h = whead) != null && (c = h.cowait) != null && U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&(w = c.thread) != null：如果头节点不为空并且头节点挂载着读模式的线程节点,当前线程则会尝试唤醒挂载的线程节点,一次只能唤醒一个线程节点。
6.2.h == (pp = p.prev) || h == p || pp == null：如果当前线程节点的上一个节点或上上个节点是头节点的话,当前线程就会尝试获取读锁。
6.3.whead == h && p.prev == pp：校验队列中的节点是否变更,如果节点没有变更则会根据后续的if语句来校验当前线程节点是重新尝试获取锁还是将它从队列中移除还是将它挂起,如果队列中的节点变更了则重新从序号1开始执行。
6.3.1.pp == null || h == p || p.status > 0：当前线程节点是头节点或者说上一个节点是头节点或者上一个节点已经取消了加锁,当前加锁的线程节点排在了对头,此时就可以将node节点置空重新走外面的大循环来尝试加锁。 此处为什么要将node节点置空而不是将该node节点从队列中移除？其实多看几遍代码和流程你就会发现,当前这个node根本就没有进入队列中,比如我们队列为空的时候会先创建一个写模式的节点,此时会将创建的节点设置为队列的头节点和尾节点,然后再为我们当前线程创建一个读模式的节点,并将这个节点的prev指针指向了刚才创建的头尾节点,可以看出这里只是我们当前线程的节点的指针指向了头尾节点,而头尾节点的指针并没有指向我们当前线程节点,所以我们当前线程的节点并没有入队,从代码上可以看出如果当前线程节点入队了其实就已经break出去了,并不会执行到当前代码这里。
6.4.如果设置了超时时间并且当前线程获取锁已经超时则会调用cancelWaiter方法将当前线程节点从队列中移除,如果没有超时或没有设置超时时间此时会将当前线程挂起等待其它线程将当前线程唤醒。
此时第一个for循环中的代码已经学习完了,现在来看一下第二个for循环中的代码,在看第二个for循环的代码前先看一下在什么情况下会从第一个for循环中退出,从第一个循环中退出分为两种情况:
- 队列中有线程节点在等待的情况下为当前线程创建了一个节点并且队列尾部的节点不是读模式的节点并且没有新的节点入队时,此时就会将当前线程节点设置为尾节点并退出循环。
- 队列中没有线程节点在等待的情况下当前线程会先创建一个写模式的节点并将该节点设置为头节点和尾节点,然后再为自己创建一个节点,如果当前线程在创建节点的过程中队列没有新的线程节点入队时,此时就会将自己的节点设置为尾节点并退出循环。
7.(h = whead) == p：校验上一个节点是否是头节点,如果是头节点那当前线程节点可以自旋尝试获取锁,如果不是那当前线程需要挂起等待其它线程唤醒。
7.1.spins < 0：在第一次进入循环的时候spins肯定是小于0的,尝试就需要根据处理器(cpu)的数量来计算自旋的次数。
7.2.spins < MAX_HEAD_SPINS：每次自旋次数耗尽时,如果需要重新自旋获取锁的时候都会校验一下是否小于最大的自旋次数,如果小于则下次自旋次数翻倍,如果不小于则说明已经到达了最大的自旋次数,后面就使用最大的自旋次数来获取锁。
7.3.for(int k = spins; ; )：自旋获取锁,如果获取锁失败,自旋次数减1,当前自旋次数耗尽时则退出自旋,根据后续的条件来决定是将当前线程节点挂起还是继续自旋,如果获取锁成功则会校验当前线程节点下是否挂载了线程节点,如果挂载了线程节点那将挂载的线程节点全部唤醒。
8.h != null：如果头节点不为空则循环校验头节点下的每一个挂载的线程节点是否为空,如果不为空则将依次将挂载的线程节点唤醒,我们知道挂载的线程节点和被挂载的线程节点都是读模式的节点,而此时头节点是一个已经获取到锁的线程节点,此时我们需要帮忙将头节点下挂载的线程节点唤醒,此时就有一个问题,既然头节点是一个获取到锁的读模式的节点,我们当前线程节点也是一个读模式的节点,那为什么在帮头节点唤醒挂载节点的时候,我们当前线程为什么不再尝试一下获取锁呢？
9.whead == h：如果头节点没有变更此时会根据后续条件来决定是否要将当前线程挂起。
9.1.(np = node.prev) != p：校验当前线程节点的上一个节点是否不等于原尾节点,如果不等于原尾节点则说明有新的线程创建了新的节点并将原尾节点的指针修改了,此时就需要将原尾节点修改成当前线程节点的上一个节点,当前线程修改了原尾节点,等后面线程执行到这里的时候会发现原尾节点与线程节点的上一个节点不相同也会进行修改。
9.2.(ps = p.status) == 0：尾节点的状态一般都是为0,只有在有新的节点入队时,此时就需要将原尾节点的状态设置为-1,状态为-1时说明后续有线程节点在等待加锁,当前节点p获取锁并释放锁之后需要唤醒唤醒节点。
9.3.ps = CANCELLED：校验上一个线程节点的状态是否为CANELLED,如果是则说明上一个线程节点已经取消了加锁,此时就需要将上一个线程节点从队列中移除。
9.4.如果设置了超时时间并且当前线程获取锁已经超时则会调用cancelWaiter方法将当前线程节点从队列中移除,如果没有超时或没有设置超时时间此时会将当前线程挂起等待其它线程将当前线程唤醒。
下图为获取读锁的整个流程,如果觉得看不清楚的话可以到把下面的链接地址复制使用浏览器打开查看 链接地址：https://www.processon.com/view/link/639d982319569d676e07e9cb
tryIncReaderOverflow
private long tryIncReaderOverflow(long s) {
// assert (s & ABITS) >= RFULL;
//校验读锁的加锁次数是否等于读锁定义的最大的加锁次数126
//此时s中代表读锁的加锁次数为126 0111 1110
if ((s & ABITS) == RFULL) {
//读锁加锁次数已到达定义的最大加锁次数
//继续尝试加读锁,如果继续加读锁成功此时state中代表加读锁的次数为127 0111 1111
if (U.compareAndSwapLong(this, STATE, s, s | RBITS)) {
//将溢出的加读锁次数使用变量存放起来
++readerOverflow;
//将s中代表读锁的加锁次数126赋予state
state = s;
//返回版本号
return s;
}
}
//有线程在加读锁或释放读锁,此时当前线程生成随机数与线程让步的概率来计算是否需要让出运行机会
else if ((LockSupport.nextSecondarySeed() & OVERFLOW_YIELD_RATE) == 0)
//让出运行机会
Thread.yield();
return 0L;
}
复制代码tryIncReaderOverflow方法会在读锁加锁次数溢出的时候会被调用,先校验读锁的加锁次数是否等于读锁定义的最大加锁次数126,如果是则通过CAS操作获取锁,获取读锁成功之后state中代表的加读锁的次数为127,此时将溢出的1次加读锁次数使用readerOverflow变量存放,并将s赋值给state并返回版本号,此时state中加读锁的次数为126,如果读锁的加锁次数不等于读锁定义的最大加锁次数,则说明有线程在加读锁或释放读锁,此时当前线程需要生成一个随机数与线程让步的概率进行计算是否需要让出运行机会。
为什么读锁定义的最大的加锁次数不是127而是126？
如果使用127来作为最大的读锁加锁次数,你会发现在多线程并发下,每个线程进入到当前方法时s都为127,此时线程A执行了CAS操作将state中加读锁的次数设置为了127,原先state中加读锁的次数就是127,跟没改一样,此时线程A将readerOverflow在主内存中的值拷贝到了自己的工作内存中准备执行++操作时,此时线程B执行了CAS操作并将readerOverflow在主内存中的值拷贝到了自己的工作内存中并对工作内存中的值进行了++操作,线程B对工作内存中的值进行了++操作后将操作后的值刷到主内存中,此时线程A对工作内存中的值进行了++操作并刷到了主内存中,此时线程A修改后的值会将线程B修改后的值覆盖,这样就会导致少一次加锁的次数,最大加锁次数为126时,每个线程进入到当前方法时s都为126,此时线程A执行了CAS操作将state中加读锁的次数设置为了127,其它线程执行CAS操作时发现state最新的值与自己预期的值不相同则会退出当前加读锁的方法等待下一次加锁。
unlockRead(释放读锁)
public void unlockRead(long stamp) {
//s 锁的状态
//m 是否有线程加了锁,加的是读锁还是写锁
long s, m;
WNode h;
for (; ; ) {
//先校验锁的版本号是否跟传递进来的版本号相同
//如果版本号相同校验是否有线程加了锁
//如果有线程加了锁则校验是否加的是写锁
if (((s = state) & SBITS) != (stamp & SBITS) ||
(stamp & ABITS) == 0L || (m = s & ABITS) == 0L || m == WBIT)
throw new IllegalMonitorStateException();
//校验加的读锁次数是否溢出
if (m < RFULL) {
//读锁次数未溢出则释放锁
if (U.compareAndSwapLong(this, STATE, s, s - RUNIT)) {
if (m == RUNIT && (h = whead) != null && h.status != 0)
//只有当前一个线程加了读锁
//并且队列中是有线程节点在等待
//当前线程释放了读锁之后则需要唤醒队列中等待加锁的线程节点
release(h);
break;
}
}
//读锁加锁次数溢出则释放锁并修改溢出的变量
else if (tryDecReaderOverflow(s) != 0L)
break;
}
}
复制代码整个释放读锁的过程都被for循环包裹着,要么释放成功要么出现异常。
((s = state) & SBITS) != (stamp & SBITS)：校验state中的版本号是否与传递进来的锁的版本号相同。(stamp & ABITS) == 0L：校验传递进来的版本号是否是一个加锁的状态(m = s & ABITS) == 0L：校验state是否是一个加锁的状态m == WBIT：校验是否是加的写锁
上述条件只要有一个成立就会抛出异常,只有都不成立的时候才会继续往下执行,校验当前加读锁的次数是否溢出,没有溢出的情况下会释放1次锁,释放完锁之后会校验是否只有当前一个线程加了1次读锁,如果是只有当前线程加了1次读锁并且队列中还有线程节点在等待加锁，那当前线程则会去唤醒队列中加锁的线程节点,如果加读锁的次数溢出则会释放1次锁并修改溢出的变量。
release(唤醒后续节点)
private void release(WNode h) {
//校验头节点是否为空
if (h != null) {
//等待唤醒的线程节点
WNode q;
//等待唤醒的线程
Thread w;
//修改头节点的状态
U.compareAndSwapInt(h, WSTATUS, WAITING, 0);
//如果头节点的下一个节点为空或已被取消加锁
//则从尾节点开始向头节点遍历获取距离头节点最近的节点状态为等待状态的节点
if ((q = h.next) == null || q.status == CANCELLED) {
for (WNode t = wtail; t != null && t != h; t = t.prev)
if (t.status <= 0)
q = t;
}
//如果节点不为空并且节点中的线程也不为空则唤醒该线程
if (q != null && (w = q.thread) != null)
U.unpark(w);
}
}
复制代码先修改头节点的状态,原先头节点的状态为-1说明头节点的后续节点需要唤醒,将头节点的状态设置为了0说明头节点后续节点唤醒之后就不需要再唤醒了,一般来说唤醒的节点是头节点的后续的一个节点,除非后续的一个节点为空或取消了加锁,此时就需要从尾节点开始向头节点遍历,获取距离头节点最近的节点状态为1的节点并将该节点唤醒。
tryDecReaderOverflow
private long tryDecReaderOverflow(long s) {
// assert (s & ABITS) >= RFULL;
//校验加锁次数是否溢出
if ((s & ABITS) == RFULL) {
//释放锁
if (U.compareAndSwapLong(this, STATE, s, s | RBITS)) {
int r;
long next;
if ((r = readerOverflow) > 0) {
//读锁的溢出次数大于0则将溢出次数减1
//而读锁的版本号还是s
readerOverflow = r - 1;
next = s;
}
else
//读锁的溢出次数等于0则说明读锁的加锁次数在126次
//并没有溢出也没有使用readerOverflow变量
//此时只需要在当前锁的版本号上减1
next = s - RUNIT;
//更新锁状态
state = next;
return next;
}
}
//有线程在加读锁或释放读锁,此时当前线程生成随机数与线程让步的概率来计算是否需要让出运行机会
else if ((LockSupport.nextSecondarySeed() & OVERFLOW_YIELD_RATE) == 0)
//让出运行机会
Thread.yield();
return 0L;
}
复制代码先校验读锁的加锁次数是否等于定义的最大读锁加锁次数,不是则说明有线程加读锁或释放读锁,此时当前线程会生成随机数与线程让步的概率来计算是否需要让出运行机会,让出了运行机会后会重新尝试释放锁直到锁释放成功,如果读锁的加锁次数等于定义的最大读锁加锁次数,先释放锁,再校验读锁加锁溢出的次数是否大于0,如果大于0则说明已经溢出,此时需要将溢出的次数-1,如果不大于0则说明没有溢出,只需要释放锁即可。
writeLock(获取写锁)
public long writeLock() {
long s, next;
return (( ((s = state) & ABITS) == 0L && U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
next : acquireWrite(false, 0L));
}
复制代码其实能看出获取写锁的方法其实与获取读锁的方法相似,主要不同的地方就是acquireWrite方法,再就是获取读锁的时候多了一个校验whead == wtail,为什么在获取读锁的时候需要先校验一下队列中的头节点是否和尾节点相等呢?其实就是为了防止获取写锁的线程饥饿的问题,如果没有该校验的话,假如线程A获取了读锁,此时获取写锁的线程B就需要进入队列中等待,此时如果后续有许多加读锁的线程来获取锁就会造成线程B一直等待,有了该校验时,后续加读锁的线程发现队列中有线程在等待,后续线程则会入队进行等待,先让队头的线程节点获取锁。
acquireWrite
private long acquireWrite(boolean interruptible, long deadline) {
//p 当前线程节点未入队时的尾节点
//node 当前线程的节点
WNode node = null, p;
for (int spins = -1; ; ) {
//s 锁状态
//m 是否已经有线程加了锁 大于1是 0否
//ns 线程加锁的版本号
long m, s, ns;
//使用锁状态与读写锁的标识进行与运算并校验是否有线程加了锁
if ((m = (s = state) & ABITS) == 0L) {
//没有线程加锁,当前线程则尝试加写锁
if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))
//加写锁成功,返回写锁的版本号
return ns;
}
//校验自旋次数是否小于0
else if (spins < 0)
//m == WBIT 校验加的锁是否是写锁
//wtail == whead 队列中的头节点是否与尾节点相同
//有线程加了写锁并且等待队列中并没有线程节点在等待或队列未被初始化则根据cpu数量来计算自旋次数
//有线程加了写锁但是队列中有线程节点在等待,那自旋次数为0,当前线程不自旋,直接创建节点并入队进行等待
//有线程加了读锁,那自旋次数为0,当前线程不自旋,直接创建节点并入队进行等待
spins = (m == WBIT && wtail == whead) ? SPINS : 0;
else if (spins > 0) {
//进入当前判断语句说明有线程加了写锁并且队列中没有线程节点在等待
//校验生成的随机数是否大于等于0
if (LockSupport.nextSecondarySeed() >= 0)
//自旋次数减1
--spins;
}
//自旋次数结束还未获取到写锁则校验队列中的尾节点是否为空
//如果队列中的尾节点为空则会初始化节点
else if ((p = wtail) == null) {
//创建一个写模式的节点
WNode hd = new WNode(WMODE, null);
//将创建的节点设置为头节点和尾节点
if (U.compareAndSwapObject(this, WHEAD, null, hd))
wtail = hd;
}
else if (node == null)
//创建一个写模式的节点,并将该节点的上一个节点的指针指向尾节点
node = new WNode(WMODE, p);
//校验当前节点的上一个节点是否是尾节点
//如果不是尾节点则说明当前线程在自旋获取写锁的时候,有其它线程来尝试获取锁并将尾节点修改了
else if (node.prev != p)
node.prev = p;
//将新创建的写模式的节点设置为尾节点,并将原尾节点的下一个节点的指针指向node
else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
p.next = node;
break;
}
}
for (int spins = -1; ; ) {
//h 头节点
//np 当前线程节点的上一个节点
WNode h, np, pp;
//ps 当前线程节点未入队时的尾节点的状态
int ps;
//校验当前线程的上一个节点是否是头节点
if ((h = whead) == p) {
if (spins < 0)
//第一次循环spins肯定是小于0的
//根据cpu数量来计算第一次入队自旋的次数
spins = HEAD_SPINS;
else if (spins < MAX_HEAD_SPINS)
//如果自旋之后还未获取到锁则将自旋的次数翻倍继续自旋
spins <<= 1;
//自旋
for (int k = spins; ; ) {
//s 锁状态
//ns 加写锁成功后的版本号
long s, ns;
//使用锁状态与读写锁的二进制位标识进行与运算获取到是否有线程加了锁
if (((s = state) & ABITS) == 0L) {
//没有线程加锁当前线程则尝试获取写锁
if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT)) {
//获取写锁成功则将当前线程节点设置为头节点
whead = node;
//取消与上一个节点的关联
node.prev = null;
//返回版本号
return ns;
}
}
//自旋次数自减
else if (LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
break;
}
}
//校验头节点是否不为空
else if (h != null) {
//读节点链表
WNode c;
//读节点线程
Thread w;
//校验读节点链表是否不为空
while ((c = h.cowait) != null) {
//如果读节点链表不为空则说明头节点是个读节点则需要将挂载在头节点上的所有读节点都唤醒
if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) && (w = c.thread) != null)
U.unpark(w);
}
}
//校验头节点是否改变
if (whead == h) {
if ((np = node.prev) != p) {
//原尾节点与当前线程节点的上一个节点不相同
//说明之前或后面来了新的线程并创建了新的节点将原尾节点修改了
//此时就需要将原尾节点修改成当前线程节点的上一个节点
//当前线程修改了原尾节点,等后面线程执行的时候会发现原尾节点与线程节点的上一个节点不相同也会进行修改
if (np != null)
(p = np).next = node;
}
else if ((ps = p.status) == 0)
//将上一个节点的状态设置为等待状态
U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
else if (ps == CANCELLED) {
//如果上一个节点的状态为取消状态则需要将当前线程节点与上一个节点取消关联
//将当前线程的节点与上一个节点的上一个节点进行关联
if ((pp = p.prev) != null) {
node.prev = pp;
pp.next = node;
}
}
else {
long time;
if (deadline == 0L)
time = 0L;
else if ((time = deadline - System.nanoTime()) <= 0L)
//已经超时
return cancelWaiter(node, node, false);
//获取当前线程
Thread wt = Thread.currentThread();
//将Thread类中的阻塞对象修改为当前类
U.putObject(wt, PARKBLOCKER, this);
node.thread = wt;
if (p.status < 0 && (p != h || (state & ABITS) != 0L) && whead == h && node.prev == p)
//挂起线程
U.park(false, time);
//清除节点中的线程引用
node.thread = null;
//将Thread类中的阻塞对象置空
U.putObject(wt, PARKBLOCKER, null);
//线程是否被中断
if (interruptible && Thread.interrupted())
return cancelWaiter(node, node, true);
}
}
}
}
复制代码acquireWrite方法中也是分为两个大的for循环,第二个for循环其实与读锁中的for循环相同,主要还是看一下第一个循环。
1.(m = (s = state) & ABITS) == 0L：先校验是否有线程加了锁,如果有则执行后续的if语句,如果没有则尝试获取写锁,获取写锁成功则返回锁的版本号。
2.spins < 0：自旋次数是否小于0,如果自旋次数小于0则会根据情况来决定自旋的次数。
- 有线程加了写锁但是队列中没有线程节点在等待,那就根据
cpu的数量来决定自旋的次数。 - 有线程加了写锁但是队列中有线程节点在等待,那自旋次数为0,当前线程不自旋,直接创建节点并入队进行等待。
- 有线程加了读锁,那自旋次数为0,当前线程不自旋,直接创建节点并入队进行等待。
3.spins > 0：自旋次数大于0则说明当前线程在自旋加锁,此时自旋次数需要-1。
4.(p = wtail ) == null ：当自旋加锁失败时会校验队列中是否有线程节点在等待,如果没有线程节点在等待则创建一个写模式的节点并将该节点设置为头节点和尾节点。
5.node == null：当前线程自旋加锁失败时会为自己创建一个写模式的节点,并将该节点的prev指针指向队列中的尾节点。
6.node.prev != p：校验当前线程在创建节点的时候是否有其它线程创建了节点并入队,导致当前线程节点指向的尾节点并不是最新的尾节点,如果不是最新的尾节点则需要指向最新的尾节点。
7.U.compareAndSwapObject(this, WTAIL, p, node)：自旋获取写锁失败之后并创建了节点,此时就需要将节点添加到队列中。
第一个for循环中的代码比较简单,大概的意思就是先看是否有线程加了锁,如果没有线程加锁,那当前线程就直接获取写锁并返回版本号,如果有则根据情况来获取自旋的次数,每自旋一次就校验锁的状态,当自旋次数耗尽并且还没获取到写锁,那就为当前线程创建一个节点并入队进行等待,第二个for的意思可以看读锁的那部分,也可以自己看一下代码理解一下。
unlockWrite(释放写锁)
public void unlockWrite(long stamp) {
//头节点
WNode h;
//校验当前释放锁的版本号是否与加锁时的版本号相同
//或者版本号加的锁不是写锁
if (state != stamp || (stamp & WBIT) == 0L)
throw new IllegalMonitorStateException();
//释放锁
//计算下一次加锁的版本号以及修改锁的状态,如果版本号为0则使用初始的版本号256
state = (stamp += WBIT) == 0L ? ORIGIN : stamp;
//校验头节点是否为空并且头节点的状态是否不为0
//如果头节点不为空并且状态不为0则说明头节点的锁释放后需要将后续线程节点唤醒
if ((h = whead) != null && h.status != 0)
//唤醒下一个线程节点
release(h);
}
复制代码先校验释放写锁的版本号是否与加锁时的版本号相同,如果不相同则说明释放写锁的版本号是一个错误的版本号,此时就需要抛出异常,如果相同则通过stamp+=WBIT来释放锁并获取到下一次加锁的版本号,如果下一次加锁的版本号等于0则说明版本号已经到达了最大值,则需要将版本号从256开始,如果队列中有线程节点在等待那释放完写锁之后就需要将队列中的线程节点唤醒。
乐观读
乐观读其实就是认为其它线程不会对共享变量进行修改,从而不加锁的去获取共享变量,我们来看一下StampedLock是如何实现乐观读的,其实很简单,先获取当前锁的版本号,然后执行代码逻辑,执行完代码逻辑之后会校验一下在执行代码逻辑期间是否有线程加了写锁,对共享变量修改了,如果有那就获取读锁,然后对需要执行的代码逻辑重新执行一遍。
public long tryOptimisticRead() {
//锁状态以及版本号
long s;
//根据锁状态返回当前是否有线程加了写锁
//!=0 有线程加了写锁
//0 没有线程加写锁
return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
}
public boolean validate(long stamp) {
//加入内存屏障,刷新数据
U.loadFence();
//计算传递进来的版本号与最新的版本号是否加了写锁
return (stamp & SBITS) == (state & SBITS);
}
复制代码通过tryOptimisticRead方法获取当前锁的版本号,如果在这个时候已经有线程加了写锁的话则返回0,当执行完代码逻辑之后调用validate方法来校验的时候会发现传递进来的锁的版本号与最新的state中的锁版本号并不相等,说明有线程修改了共享变量,此时就需要重新执行代码逻辑。
public static void main(String[] args) {
//创建锁对象
StampedLock stampedLock = new StampedLock();
//获取锁的版本号
long tryOptimisticRead = stampedLock.tryOptimisticRead();
//具体逻辑
System.out.println(number);
//校验是否有其它线程在当前线程执行具体逻辑的时候加了写锁
if (!stampedLock.validate(tryOptimisticRead)) {
//有其它线程加了写锁,当前线程则获取读锁
long readLock = stampedLock.readLock();
//具体逻辑
System.out.println(number);
//释放读锁
stampedLock.unlockRead(readLock);
}
}
复制代码锁转换
tryConvertToReadLock：将锁转换为读锁,如果本身就是一个读锁并不会改变,如果你传递的版本号是一个没有加读锁的版本号并且当前也没有线程加锁,此时当前方法就会获取一个读锁并返回版本号,主要还是将写锁转换为读锁,其实底层就是先释放写锁再获取读锁。tryConvertToWriteLock：将锁转换为写锁,如果本身就是一个写锁并不会改变,如果你传递的版本号是一个没有加写锁的版本号并且当前也没有线程加锁,此时当前方法就会获取一个写锁并返回版本号,读锁想要转换为写锁,必须只有当前一个加读锁的线程。tryConvertToOptimisticRead：将锁转换为乐观读,其实就是不管你是写锁还是读锁都给释放掉。
