首先声明一点: 我在分析源码的时候, 把jdk源码复制出来进行中文的注释, 有时还进行编译调试什么的, 为了避免和jdk原生的类混淆, 我在类前面加了"My". 比如把ReentrantLock改名为了MyReentrantLock, 在源码分析的章节里, 我基本不会对源码进行修改, 所以请忽视这个"My"即可.
一. 简介
锁是什么? 锁是一种标志, 或者是一种资源, 持有锁的线程才可以继续往下执行相应的内容. 未持有锁的线程需要等待这个锁资源. 直到获取到了这个锁, 才可以继续向下执行.
0. ReentrantLock的一个小demo
想自己运行这段代码的话, 把代码中的"MyReentrantLock" 改为 "ReentrantLock" 即可. (后续的代码也一样, 如果想自己运行, 还编译报错, 请把我修改的代码改回来. 也就是把"My"都去掉就好了)
public class Main { private static MyReentrantLock lock = new MyReentrantLock(); public static void main(String[] args) throws Exception { // 场景如下: 线程1先获得锁, 释放后, 线程2 再获得锁. new Thread(() -> { System.out.println("线程1启动"); lock.lock(); System.out.println("线程1抢到锁"); try { System.out.println("这里是业务逻辑1"); quietSleep(2);// 两秒后释放锁 System.out.println("两秒后"); } finally { lock.unlock(); System.out.println("线程1释放锁"); } }).start(); new Thread(() -> { System.out.println("线程2启动"); quietSleep(1); // 在这里进行谦让. 确保上面的线程能先运行. 也就是让上面的线程先获得锁 lock.lock(); System.out.println("线程2抢到锁"); try { System.out.println("这里是业务逻辑2"); } finally { lock.unlock(); System.out.println("线程2释放锁"); } }).start(); } public static void quietSleep(long sec) { try { Thread.sleep(sec * 1000); } catch (InterruptedException e) { e.printStackTrace(); } } }
输出的结果如下:
1. sync字段
首先来看一下ReentrantLock里唯一的一个字段
Sync继承自AQS(AbstractQueuedSynchronizer, 以下简称AQS) . 公平锁和非公平锁都继承了Sync. Sync是ReentrantLock类里锁的统一声明.
2. lock/unlock依赖Sync
ReentraintLock的 lock()和unlock()方法实际上都是靠Sync来实现的:
3. 锁内部类定义
Sync 和 公平锁 和 非公平锁 都是ReentrantLock的内部类, 类的定义部分如下(细节先隐藏起来了, 后面会讲):
4. ReentrantLock构造器
ReentrantLock有两个构造器.
1. 默认构造器是直接使用了非公平锁. 非公平锁就是不一定按照"先来后到"的顺序来进行争抢.
2. 带参构造器可以传递一个bool类型. true的时候为公平锁. 公平锁就是按照"先来后到"的顺序来进行争抢.
二. 公平锁申请锁
使用锁的第一个步骤, 当然就是先申请锁了, 咱么来分析一下源码, 看看申请锁的流程吧.
1. 公平锁获取锁的流程(单线程, 没有争抢)
首先从最外层的调用lock()方法开始. 咱们在Main方法里写下这两行代码:
MyReentrantLock就是ReentrantLock, 我复制了源代码, 然后改了个名字而已.
Reentraint类的lock()方法最终还是调用的sync.lock()
由于我们现在使用的是公平锁. 所以sync现在是FairSync. 所以sync.lockI()实际上就是FairSync类里的lock()方法
发现lock()调用的是acquire(1)这个方法, 这个方法是在AQS类里实现的.代码如下:
arg当时传进来的是1, 所以首先进行的是tryAcquire(1)来进行"尝试获取锁"的操作. 这时一种乐观的想法.
tryAcquire方法的具体实现在FairSync类里, 具体代码如下:
/** * @return 返回true: 获取到锁; 返回false: 未获取到锁 * 什么时候返回true呢? 1.没有线程在等待锁;2.重入锁,线程本来就持有锁,也就可以理所当然可以直接获取 * @implNote 尝试直接获取锁. */ protected final boolean tryAcquire(int acquires) { // 获取当前线程的引用 final Thread current = Thread.currentThread(); // 当前锁的计数器. 用于计算锁被获取的次数.在重入锁中表示锁重入的次数.由于这个锁是第一次被获取, 所以c==0 int c = getState(); // c==0, 也就是 state == 0 ,重入次数是0, 表示此时没有线程持有锁. if (c == 0) { // 公平锁, 所以要讲究先来后到 // 因为有可能是上一个持有锁的线程刚刚释放锁, 队列里的线程还没来得及争抢, 本线程就乱入了 // 所以每次公平锁抢锁之前, 都要判断一下等待队列里是否有其他线程 if (!hasQueuedPredecessors() && // 执行到这里说明等待队列里没有其他线程在等待. // 如果没有线程在等待,那就用CAS尝试一下,成功了就获取到锁了, // 不成功的话,只能说明一个问题,就在刚刚几乎同一时刻有个线程抢先了 =_= compareAndSetState(0, acquires)) { // 到这里就获取到锁了,标记一下,告诉大家,现在是我(当前线程)占用了锁 setExclusiveOwnerThread(current); // 成功获取锁了, 所以返回true return true; } //-- 由于现在模拟的是单纯地获取一次锁, 没有重入和争抢的情况, 所以执行不到这里, 上面的cas肯定会成功, 然后返回true } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }
争抢完锁之后会返回true, 然后回到上层方法acquire :
if语句里 && 前面是false, 不会继续往下执行了. 当前线程获取到了锁, 而且执行了所有该执行的内容, 就完事儿了.
2. 公平锁进行重入的流程
重入就是一个线程获取到了锁, 然后这个线程又一次申请(进入)了这个锁.
重入用synchronized来举例就是这样:
用ReentrantLock来举例子就是这样:
同一个线程(main线程) 首先进行了lock.lock()申请并占有了锁, 随后又执行了一次lock.lock(). 还没释放锁的情况下, 又一次申请锁. 这样就是重入了.
上面一小节已经分析了第一行的lock.lock()是如何获取到锁的, 所以我们只分析 重入的部分, 也就是后面那句lock.lock()的执行流程.
前面的执行过程一直是一模一样的, 直到这里:
/** * @return 返回true: 获取到锁; 返回false: 未获取到锁 * 什么时候返回true呢? 1.没有线程在等待锁;2.重入锁,线程本来就持有锁,也就可以理所当然可以直接获取 * @implNote 尝试直接获取锁. */ protected final boolean tryAcquire(int acquires) { // 获取当前线程的引用 final Thread current = Thread.currentThread(); // 当前锁的计数器. 由于前面的那句lock已经获取到锁了, 所以这里是status==1, 也就是 c==1 int c = getState(); // c==1, 表示当前有线程持有锁, 所以这段if是进不去了 if (c == 0) { if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } // 由于 c==1 , 无法进入if语句, 所以来看看满不满足这里的 else if // 这个锁被人占了, 但还是不死心, 于是看一下是不是当前线程自己占的这个锁. // (人家女生说有喜欢的人, 为什么不问问是不是自己呢 = =.) // 由于是同一个线程, 所以就是自己啦! 所以会进入这个else if分支, } else if (current == getExclusiveOwnerThread()) { // 代码执行到这里了, 就是所谓的 重入 了 // 这里的acquires的值是1, 所以nextc = 1 + 1 , 也就是2了 int nextc = c + acquires; // 小于0, 说明int溢出了 if (nextc < 0) throw new Error("Maximum lock count exceeded"); // 在这里把状态更新一下, 把state更新为2, 意思就是这个锁被同一个线程获得2次了. // (大家就可以以此类推, 下次再重入的话, 那么就会再+1, 就会变为3....) setState(nextc); // 重入完成, 返回true return true; } return false; }
还记得上小节讲的, 获取锁的时候进入的是这段代码的if语句, 而重入就不一样了, 进入的是 else if语句. 但最终返回的还是true, 表示成功.
上面讲的是无争强的情况, 接下来讲讲有争抢的情况.
3. 公平锁cas争抢失败
场景如下:
一开始锁是空闲状态, 然后两个线程同时争抢这把锁(在cas操作处发生了争抢).
一个线程cas操作成功, 抢到了锁; 另一个线程cas失败.
代码例子如下(代码的意思到位了, 但是这段代码最后不一定会在cas处进行争抢, 大家意会就好了):
cas操作成功的线程就和上面第1小节的一样, 就不用再重复描述了.
而cas争抢失败的线程会何去何从呢? 看我给大家分析:
/** * @return 返回true: 获取到锁; 返回false: 未获取到锁 * 什么时候返回true呢? 1.没有线程在等待锁;2.重入锁,线程本来就持有锁,也就可以理所当然可以直接获取 * @implNote 尝试直接获取锁. */ protected final boolean tryAcquire(int acquires) { // 获取当前线程的引用 final Thread current = Thread.currentThread(); // 当前锁的计数器. int c = getState(); // state == 0 表示此时没有线程持有锁 if (c == 0) { // 本场景中, 一开始锁是空闲的, 所以队列里没有等待的线程 if (!hasQueuedPredecessors() && // 两个线程在这里进行争抢 // cas抢成功的会进入到if代码块 // cas抢失败的, 就跳出整个if-else, 也就是直接到最后一行代码 compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } // cas 操作失败后, 会这直接执行到这里. 返回false. return false; }
在这里返回了false, 回到上一层函数.
第一个条件是true, 所以会继续往下执行acquireQueued方法. 来准备让这个失败的线程进入队列等待.
下面继续来给大家讲解 acquireQueued(addWaiter(Node.EXCLUSIVE), arg) .
先讲讲这个addWaiter(Node.EXCLUSIVE):
/** * 将当前线程封装为Node, 然后根据所给的模式, 进行入队操作 * * @param mode 有两种模式 Node.EXCLUSIVE 独占模式, Node.SHARED 共享模式 * @return 返回新节点, 这个新节点封装了当前线程. */ private Node addWaiter(Node mode) { // 这个mode没用上. Node node = new Node(Thread.currentThread(), mode); // 咱们刚才都没见到过tail被赋予了其他的值, 当然就是null了. Node pred = tail; // tail是null的话, pred就是null, 所以不会进入到这个if语句中.所以跳过这个if语句. if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 因为锁的等待队列是懒初始化, 直到有节点插入进来, 它才初始化. // 而现在这个挣钱失败的线程, 正好是锁建立以来, 第一个进入等待队列的线程. 所以现在才准备进行初始化. // 初始化完了后会把当前线程的相关信息和引用封装成Node节点, 然后插入到队列当中.并且制定head 和 tail. // tail就不等于null了, 所以下一次addWaiter方法被调用的时候, 就会执行上面的if语句了. 而不会跳过if语句, 来到这里进行初始化了. enq(node); // 返回这个Node节点. return node; }
目的就是要将这个cas失败的线程封装成节点, 然后插入到队尾中. (等待队列是懒初始化,)
如果队列已经初始化了, 那么tail就不会是null, 就会执行上面代码中的if语句, 调整一下指针的引用就好了.
但是如果队列还未初始化, 那么就应该先初始化, 再插入. 先初始化,再插入, 对应的代码是enq(node).
接下来讲解一下enq方法:
/** * 采用自旋的方式入队 * CAS设置tail,直到争抢成功. */ private Node enq(final Node node) { for (; ; ) { Node t = tail; // 最开始tail肯定是null, 进入if进行初始化head和tail. if (t == null) { // Must initialize // 设置head 和tail. cas来防止并发. if (compareAndSetHead(new Node())) tail = head; // if 语句执行完了后, 之后的for循环就会走else了. } else { // 争抢入队, 没抢到就继续for循环迭代.抢成功了就可以return了,不然一直循环. // 为什么是用cas来争抢呢? 因为怕是多个线程一起执行到这里啊 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
刚才的addWaiter(Node.EXCLUSIVE) 分析完了, 总之就是addWaiter之后, 队列肯定是被创建完了, 而且还把node(当前线程的封装)插入到了队列的队尾. 并且返回了这个node. acquireQueued(addWaiter(Node.EXCLUSIVE), arg) 可以简化为 acquireQueued(node)
所以继续分析acquireQueued方法.
final boolean acquireQueued(final Node node, int arg) { // node是刚才addWaiter方法插入到队尾的节点 // arg 是 1 boolean failed = true; try { boolean interrupted = false; for (; ; ) { // 获取node节点的前驱. final Node p = node.predecessor(); // 如果node节点的前驱是head if (p == head // 那么可以再尝试着抢一下锁. // 等待队列里的第一个节点很乐观, 因为确实很有可能会马上轮到他 && tryAcquire(arg)) { // 如果这个node就是那么巧合, 刚刚锁被释放了, 这回重新抢就真的抢到了 // 那么就把当前节点设为头结点.(头结点的含义就是当前持有锁的线程) setHead(node); // 上一个节点既然已经释放了锁, 也就该GC了. 置为null, 方便GC收集 p.next = null; // help GC // 很明显是获取锁成功了啊, 所以failed = false failed = false; // 这么大一段代码, 只有这一处return return interrupted; } //---- 如果不是队头, 那么就会执行到这里. //---- 或者虽然作为等待队列里的第一名, 单由于持有锁的线程还是没有释放, 所以还是没抢到锁. 那么也会执行到这里 // 获取锁失败的时候是否该阻塞 if (shouldParkAfterFailedAcquire(p, node) // 在这里阻塞, 等待唤醒 && parkAndCheckInterrupt()) interrupted = true; } } finally { // 上面那段, 如果中途异常了的话, 就会执行到这里. (一般不会到这里的) if (failed) cancelAcquire(node); } }
上面这段代码中shouldParkAfterFailedAcquire方法 和 parkAndCheckInterrupt() 方法 还未解释. 一个一个来.
/** * 当前线程没有抢到锁,是否需要挂起当前线程 * * @param pred 前驱结点 * @param node 当前结点 * @return 如果线程需要被阻塞, 那么就返回true */ private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; // 前驱节点的 waitStatus == -1 ,说明前驱节点状态正常,当前线程需要挂起,直接可以返回true if (ws == Node.SIGNAL) return true; // 大于0, 其实就是等于1, Node.CANCELLED 是 1, 因为状态中只有这个状态是大于0的...说明前驱节点取消了排队 // 所以下面这块代码说的是, 在链表中从prev结点开始, 往前删掉CANCELLED状态的结点. // 只有CANCELLED状态值大于0 if (ws > 0) { do { node.prev = pred = pred.prev; // 删掉之后再往前看看, 看看前面是不是CANCELLED, 如果是, 那还得继续往前删 } while (pred.waitStatus > 0); pred.next = node; } else { // 在前面的两个if语句中排除掉了waitStatus值为-1和1的情况, // 只剩下0,-2,-3这三个状态了 // 然而在我们前面的源码中,都没有看到有设置waitStatus的, // 所以只剩下等于0的情况了 // 下面的操作就是, 如果waitStatus等于0, 那么就用cas将前驱结点的waitStatus设置为-1 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
shouldParkAfterFailedAcquire里的前两行是在判断前驱节点prev的状态. 但是之前咱们分析代码, 并没有发现哪里设置了waitState.
所以waitState是默认值0.
所以shouldParkAfterFailedAcquire会直接执行下面的else, 在这里吧pred的waitState设置为-1, 然后返回false.
回到刚才的acquireQueued方法. 由于外层是for循环, 会在下一次for循环在此执行到shouldParkAfterFailedAcquire方法.
由于刚才已经把前驱节点prev的waitState改为1了, 所以这次在前两行判断prev的waitState时, 直接就满足条件, 然后return true了.
shouldParkAfterFailedAcquire方法return true了, 才会往下执行parkAndCheckInterrupt方法.
下面是parkAndCheckInterrupt()方法. 最终返回Thread.interrupted(). 返回线程是否被中断. (中断和挂起不是一回事 )
/** * 在这里线程阻塞. * 被唤醒的时候会返回, 如果被中断过, 那么就返回true * * @return {@code true} if interrupted */ private final boolean parkAndCheckInterrupt() { // 挂起. MyLockSupport.park(this); return Thread.interrupted(); }
LockSupport.park(this)会挂起当前线程. 但是LockSupport.park还有一个隐藏功能. 就是, 如果先对一个线程unpark, 再对这个线程park, 那么这次的park是失效的. 下一次park才会挂起.
原因就是, 对一个没有被park的线程进行unpark的时候, 会把标志位perm置为1. 而每次park的操作, 都是先去检查perm是否为1.
如果是1, 那么置为0, 并且这次不挂起.
如果perm为0, 那么就直接挂起这个线程.
4. 公平锁由于队列内有元素而失败
demo如下. 前两个线程, 其中一个获取锁成功, 另一个失败, 然后进入等待队列.
稍后, 第三个线程来获取锁, 但是这时由于等待队列中已经有元素在等待了. 所以会直接失败, 然后会被插入到等待队列的尾部.
上面的main方法中总共有三个线程想要占有锁. 前两个锁的争抢在上小节就已经模拟过了.
咱么现在只分析第三个线程申请锁的流程. 这个场景下的tryAcquire方法如下(会直接返回false):
/** * @return 返回true: 获取到锁; 返回false: 未获取到锁 * 什么时候返回true呢? 1.没有线程在等待锁;2.重入锁,线程本来就持有锁,也就可以理所当然可以直接获取 * @implNote 尝试直接获取锁. */ protected final boolean tryAcquire ( int acquires){ // 获取当前线程的引用 final Thread current = Thread.currentThread(); // 当前锁的计数器. int c = getState(); // 不会走这的if语句, 因为锁被其他线程占有, 肯定不是0 if (c == 0) { if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } // 由于队列内有元素, 所以if语句不执行, // 由于不是重入, else if 也不执行. // 直接返回false return false; }
这段方法返回false, 说明需要执行这个. acquireQueued(addWaiter(Node.EXCLUSIVE), arg). 先看看addWaiter方法有什么区别.
/** * 将当前线程封装为Node, 然后根据所给的模式, 进行入队操作 * * @param mode 有两种模式 Node.EXCLUSIVE 独占模式, Node.SHARED 共享模式 * @return 返回新节点, 这个新节点封装了当前线程. */ private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // 以下几行代码想把当前node加到链表的最后面去,也就是进到阻塞队列的最后 Node pred = tail; // 如果tail不是空, 说明有头结点.说明这个队列已经被初始化了. // 因为本小节讲的就是: 因为公平锁的等待队列中有其他线程才导致当前线程争锁失败, 所以说明等待队列不仅被初始化了, 而且里面还有元素. if (pred != null) { // node设置自己的前驱为pred node.prev = pred; // 用CAS把当前节点node设置为队尾, 如果成功后,tail指针就指向了node if (compareAndSetTail(pred, node)) { // 如果cas争抢成功, 那么就会在这里返回.(而cas失败的, 会跳过这个if代码块, 会执行到下面的enq方法) // 剩下的就是整理一下链表数据结构的连接问题了 // pred调整自己的后继为node pred.next = node; return node; } } // 如果在上面的cas中设置失败, 那么还是会执行到这里. // 然后在enq方法里靠for循环+cas的形式, 不断尝试着插入到队尾. enq(node); return node; }
后续执行的就和上小节的一样了.就不重复了...
当然, 场景是举不完的, 举完的话就跟笛卡尔积那样了. 我这里只是靠这四个例子来尽量完整地分析了获取锁的流程.
三. 公平锁释放锁
刚才申请锁的流程. 但是争抢失败的那些线程, 最后都进入到了等待队列里, 然后就杳无音讯了.
那当前持有锁的线程释放锁后, 是如何唤醒等待队列里的线程, 让下一个线程获取锁的呢?
咱么接下来分析一下释放锁的过程吧.
1. 申请1次锁, 执行一些业务, 然后释放
咱们只关注unlock, lock就跳过了, 前面讲过了.
ReentrantLock类的lock()方法 代码如下:
而这个release是AQS里的方法. 源码如下:
其中arg变量值是1. 首先会执行tryRelease(1) 来尝试释放锁.
如果尝试成功了, 那么tryRelease(1)就会返回true, 就会继续执行if代码块里的内容.
如果尝试失败了, 那么tryRelease(1)就会返回false. 然后就会跳过if语句, 最终本段方法(release方法)也会返回false.
咱们先分析一下tryRelease方法吧(tryRelease方法的源码在Sync抽象类里):
protected final boolean tryRelease(int releases) { // releases == 1 // c 就是重入次数 -1 , 由于本场景下模拟的是简单的获取一次锁, 然后释放, 不涉及到重入. 所以getState() == 1 // 所以c = 1 - 1 , c现在等于0 int c = getState() - releases; // 判断当前的线程是不是持有锁的线程, 不然抛异常. // 这是为了其他的线程捣乱. if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); // 用于标记是否可以完全释放锁 boolean free = false; // c等于0, 说明没有重入了, 可以完全释放了. if (c == 0) { // 标记一下, 准备完全释放锁 free = true; // 把锁的持有者设置为空, 表示锁被释放. setExclusiveOwnerThread(null); } // 把刚才c==0 设置为state setState(c); // 表示是否完全释放. 本场景下返回true return free; }
由于返回的是true, 所以返回后还有if语句块要执行:
接下来分析一下其中的unparkSuccessor方法, 看看他是如何唤醒下一个节点的.(这个方法在AQS里)
unpark之后, 就会把之前park(挂起)的线程激活, 然后继续执行:
如果线程被中断了, 那么parkAndCheckInterrupt()方法会返回true, 然后就会执行interrupted = true 这句话.
挂起和中断不是一回事, 一般不会被中断的. 所以一般不会执行interrupted=true这句话.
外层是个for循环, 当前线程被激活后, 作为等待队列中的第一个线程, 来进行获取锁. 由于是公平锁, 所以可以放心拿到, 没有人会抢, 所以会正常获取到锁.
2. 重入锁的释放
释放重入的锁(同一个线程多次获取的锁), 执行流程唯一不同的就是tryRelease方法了, 其他的都一样, 可以直接参考上面一小节的.
咱么看看重入的时候, tryRelease是如何执行的吧.
protected final boolean tryRelease(int releases) { // 其实就是重入计数器 -1 // 而由于本线程获取了2次这个锁, 所以state字段的值为2 // 所以c = 2 - 1 // 所以现在c == 1 int c = getState() - releases; // 判断当前的线程是不是持有锁的线程, 不然抛异常. if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); // 用来标记是否完全释放锁 boolean free = false; // c现在等于1, 不会进入这个if代码块 if (c == 0) { free = true; setExclusiveOwnerThread(null); } // 设置重入计数器, 也就是让 state =1 setState(c); // 返回false. return free; }
本小节和上小节的区别就是这段代码了.既然这段方法返回false, 那么返回后, release方法的if代码块自然也就不执行了.
四. 非公平锁的获取
刚才讲解了公平锁, 那么接下来讲讲非公平锁, 到底是怎么个不公平呢?
1. 非公平锁与公平锁获取的区别
由于非公平锁的获取与公平锁的获取, 只有一点点区别. 所以咱么只分析出区别就好了, 其他的部分都一样的.
然后会调用到NonfairSync类里的lock()方法.
这里就体现出了区别.
公平锁里的lock()方法里面, 只有acquire(1).
而非公平锁在acquire(1)之前多了一次cas操作. 一上来就尝试着抢占锁, 看看有没有机会(万一真的这个时候持有锁的线程正好把锁释放了呢). 非公平锁根本不管是否有其他人在排队.上来就是一抢.
当这次cas失败了, 才会像公平锁一样进入acquire(1)方法:
这里和公平锁一样. 只是, 非公平锁的tryAcquire方法和公平锁的tryAcquire方法内部实现不一样.
看看非公平锁的tryAcquire方法吧:
咱们继续往下看看nonfairTryAcquire方法吧:
/** * 不公平地尝试获取锁. * 不公平的语义就是: 不用判断队列里是否有其他线程在等待, 直接抢. */ final boolean nonfairTryAcquire(int acquires) { // 获取当前线程的引用 final Thread current = Thread.currentThread(); // 当前线程的重入次数 int c = getState(); // 如果是0, 表示此时此刻锁还被被任何一个线程所占用 if (c == 0) { // 当c==0的时候, 公平锁锁是先判断队列里是否有其他线程在等待, 如果没有, 再去cas争抢. // 而非公平锁这里, 就是根本就不去理会等待队列, 自己抓到机会就赶紧抢 // cas来争抢, 让重入次数变1. // 用cas是因为这个地方会发生并发. // 多个抢占当然只有一个成功了 if (compareAndSetState(0, acquires)) { // 设置锁的拥有者为当前线程. setExclusiveOwnerThread(current); return true; } // 如果不是0, 说明锁被某一个线程占用了 // 既然被占用了, 那就有两种情况: 1. 被自己占用; 2. 被别的线程占用 // 所以先看看是不是自己占用的, 如果是自己占用的, 那就重入. } else if (current == getExclusiveOwnerThread()) { // 其实就是+1 int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); // 这里不会产生争抢, 不必用cas // 因为只有占用锁的这一个线程才能进入到这个else if 里 // 一个线程不可能发生争抢 setState(nextc); return true; } // 1. 如果在if里的cas争抢失败 // 2. 或者是不满足else if的条件 // 那就会直接返回false // 不管是成功还是失败, 都不会有线程的等待阻塞之类的. 都是立即返回. return false; }
这里的非公平锁的nonfairTryAcquire方法 和 公平锁的tryAcquire方法很像. 区别就是:
非公平锁是, 当c==0. 也就是此时此刻, 锁是空闲状态的时候. 直接就尝试着用cas来争抢锁, 看看是否能成功, 而不管等待队列是否还有其他线程再等待.
而公平锁在c==0的时候, 也就是state==0 的时候, 先去看看队列里是否有其他线程再等待, 如果队列里没有其他线程在等待, 才会去cas争抢. 不然就会把机会让给队列里的第一个线程, 而自己会进入到等待队列的尾部.
为什么c==0了, 队列里还有可能会有其他的元素在等待呢?
因为c==0只是说明当前锁的状态是空闲状态. 只是上一个线程刚刚把锁释放, 当前线程就来争抢锁了, 还没来得及唤醒等待队列里的第一个线程呢.
其他地方就跟公平锁都一样的, 就是多了本小节讲的两处cas.
五. 非公平锁的释放
1. 非公平锁会导致饥饿
也就是说, 上一个线程释放锁后, `等待队列` 里的第一个线程就会被激活, 然后会执行tryAcquire方法. 如果这个时候有新的线程来争抢,
由于是非公平模式, 有可能新的线程会抢到这个锁. 如果新的线程抢到了锁, 那么刚刚被激活的线程(等待队列里的第一个线程)就是执行tryAcquire失败, 这个方法执行失败就意味着会被再次被挂起. 如果并发量严重, 很可能`等待队列`里的所有线程在一定时间内都无法被正常调度.也就是产生了线程饥饿的现象.
六. Condition简介
1. condition简介和demo
public class Main { private static MyReentrantLock lock = new MyReentrantLock(); private static Condition condition = lock.newCondition(); public static void main(String[] args) { new Thread(Main::funcA).start(); new Thread(Main::funcB).start(); } public static void funcA() { lock.lock(); System.out.println("await之前"); try { condition.await(); // 在这里等待被其他线程通知(signal) } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("await之后"); lock.unlock(); } public static void funcB() { lock.lock(); System.out.println("signal之前"); condition.signal(); // 这里会通知funcA的await().让funcA()继续执行下去 System.out.println("signal之后"); lock.unlock(); } }
运行这段代码, 输出如下:
如果还是没有体会到区别, 那么把main方法里的第二行注释掉, 然后再执行一下:
输出结果如下:
也就是说, await()会使当前线程挂起, 需要其他线程通知他, 他才能被激活(唤醒).
七. Condition实例化
1. 获取condition的例子
2. condition实例化的源码
newCondition方法在ReentrantLock类里的实现如下:
Sync类里的newCondition()方法如下:
ConditionObject是AQS里的一个内部类,实现自Condition, 类的声明如下(具体源码后面再解释):
八. condition的等待(await) 和 通知(signal)
1. 只执行一句await()后的流程
await()方法的具体实现在AQS里的内部类ConditionObject类里:
public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 添加到 condition 的`条件队列`中 Node node = addConditionWaiter(); // 完全释放锁,返回值是释放锁之前的 state 值 int savedState = fullyRelease(node); int interruptMode = 0; // 这里的isOnSyncQueue就是在判断node节点是否在锁的`等待队列`里 while (!isOnSyncQueue(node)) { // 在这里线程挂起 MyLockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } //---- 程序不会执行到下面, 因为在前面就已经挂起了. if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); }
这里面有几个方法之前没提到过, 在这里一一攻破.
先解决addConditionWaiter()方法:
/** * 将当前线程对应的节点入队,插入队尾, 并且作为本方法的返回值. */ private Node addConditionWaiter() { // 本例子中的场景下, 只执行过一次await()方法, 所以是第一个进入`条件队列`的元素. // 所以lastWaiter和firstWaiter肯定都是null. Node t = lastWaiter; // 本例子中t==null, 所以这段if暂时不考虑吧 if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } // 新建节点 Node node = new Node(Thread.currentThread(), Node.CONDITION); // 因为t==null, 意思是队列目前还是空的, 所以这个节点是第一个节点, 所以是firstWaiter. if (t == null) firstWaiter = node; else t.nextWaiter = node; // 但node同时也是最后一个节点, 也就是lastWaiter lastWaiter = node; // 最后会返回本方法 return node; }
接下来是fullRelease(node)方法, 来完全释放锁:
final int fullyRelease(Node node) { boolean failed = true; try { int savedState = getState(); // 这里使用了当前的 state 作为 release 的参数,也就是完全释放掉锁,将 state 置为 0 if (release(savedState)) { failed = false; // 并且把释放锁之前的state值返回出去. (本例子中是1) return savedState; } else { throw new IllegalMonitorStateException(); } } finally { if (failed) node.waitStatus = Node.CANCELLED; } }
最后就是isOnSyncQueue(node)方法, 来判断锁的`等待队列`中有没有当前这个node:
/** * 这个方法就是判断 node 是否已经移动到sync queue了 * (signal 的时候会将节点从条件队列移到sync queue) */ final boolean isOnSyncQueue(Node node) { // 当进入Condition队列时,waitStatus肯定为CONDITION, // 如果同时别的线程调用signal,Node会从Condition队列中移除, // 并且移除时会清除CONDITION状态。 // 从移除到进入sync queue队列,中间这段时间prev必然为null,所以还是返回false,即被park if (node.waitStatus == Node.CONDITION || node.prev == null) // 本例子中, 会在这里返回 return false; //--- 本例子中, 程序不会往下执行了. 但是下面的代码还是分析一下吧. 这样待会儿就不用再重新讲个方法了. // 当别的线程进入sync queue队列时,会和前一个Node建立前后关系,所以如果next存在,说明一定在release队列中 if (node.next != null) // If has successor, it must be on queue return true; // 到这里还没找到, 那只能去锁的`等待队列`里一个一个找了 // 可能该Node刚刚最后一个进入release队列,所以是tail,其next必然是null,所以需要从队尾向前查找 // 这个方法的源码就不讲了, 太简单了, 就是链表从后往前找node.找到了就true.没找到就false. return findNodeFromTail(node); }
最终会执行到await()方法里的park()方法, 线程挂起. 等待被别的线程唤醒.
2. 只执行一句signal()后的流程
然后咱们看看signal()的源码.
由于firstWaiter==null, 所以first==null, signal方法直接就退出了.
3.一个线程await等待, 另一个线程用signal来唤醒
本场景的程序demo如下:
public class Main { private static Scanner scanner = new Scanner(System.in); private static MyReentrantLock lock = new MyReentrantLock(); private static Condition condition = lock.newCondition(); public static void main(String[] args) throws Exception { new Thread(Main::funcA).start(); new Thread(Main::funcB).start(); } public static void funcA() { lock.lock(); System.out.println("await之前"); try { condition.await(); // 在这里等待被其他线程通知(signal) } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("await之后"); lock.unlock(); } public static void funcB() { lock.lock(); System.out.println("signal之前"); System.out.print("请输入任意内容并回车, 以执行signal方法: "); scanner.next(); // 在这里进行阻塞, 在控制台输入任意内容后回车, 就会接触阻塞, 就会执行signal方法, 也就是通知funcA()方法. condition.signal(); // 这里会通知funcA的await().让funcA()继续执行下去 System.out.println("signal之后"); lock.unlock(); } }
首先, await()仍然执行到park这句, 然后挂起, 这点与本章第1小节的流程是一样的(看下图, 我选中的park那行代码, await就在这里挂起):
而此时控制台如下:
此时还没有执行signal, 因为我用输入流给signal方法进行阻塞了, 需要输入内容后回车, 就可以调用到signal方法.signal通知后,await就会被唤醒.
如下:
咱们分析一下signal是如何通知await, 然后让await线程被唤醒的:
因为刚才执行过await(), 所以firstWaiter不会是null. 所以会调用到doSignal方法:
上面这段代码也比较简单, 就是将firstWaiter为头的这个链表, 把第一个元素出队, 然后让第二个元素当新的头部. 然后让刚才出队的那个元素执行tansferForSignal方法.
/** * 将节点从条件队列转移到锁的`等待队列` * * true 代表成功转移 * false 代表在 signal 之前,节点已经取消了 */ final boolean transferForSignal(Node node) { /* * 在这里将 waitStatus 置为 0. * 如果成功设置为0, 那么继续往下面执行 * 如果CAS 失败,说明此 node 的 waitStatus 已不是 Node.CONDITION,说明节点已经取消,那么直接return false. */ if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; /* * enq(node): 自旋进入阻塞队列的队尾.这个在将lock()方法的时候大家见到过.就是同一个方法. * 这里的返回值 p 是 node 在阻塞队列的前驱节点 */ Node p = enq(node); int ws = p.waitStatus; // ws > 0 说明 node 在阻塞队列中的前驱节点取消了等待锁,直接唤醒 node 对应的线程。 // 如果 ws <= 0, 那么 compareAndSetWaitStatus 将会被调用 // 因为节点入队后,需要把前驱节点的状态设为 Node.SIGNAL(-1) if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) // 如果前驱节点取消或者 CAS 失败,会进到这里唤醒线程(但是本场景下不会执行到这里) MyLockSupport.unpark(node.thread); // 返回true return true; }
将上面这段代码总结一下就是: 将本节点的waitState设置为了0. 然后让本节点插入到到锁的`等待队列`, 然后将前驱节点的waitState设置为了1. 然后返回了true.
这一行的tansferForSignal返回了true, 取反了就是false了, 所以退出了 while循环. 至此signal方法就执行完毕了.
signal干的主要事情就是: 把`条件队列`里的第一个元素转移(尾插)到了锁的`等待队列`里.
`条件队列`就是firstWaiter为头结点的一个链表.
`等待队列`就是咱们上面将lock() unlock()的时候提到的锁的等待队列.
signal方法执行完了后, 接下来就该执行unlock()方法了. 如下图:
unlock()所做的事情就是, 释放当前的锁, 然后激活`等待队列`里的第一个线程.
而在本场景下, 现在等待队列里有且仅有一个元素, 就是signal方法转移的那个元素.
unlock()之前分析过, unlock会调用release方法:
release方法所做的就是释放锁(第一个红色代码), 然后唤醒`等待队列`里的第一个线程(第二个红色代码).
unlock()方法执行完了后, 刚才await挂起的那个线程就又被激活了.
所以接下来执行的是acquireQueued方法, 这个方法在将锁的时候讲过, 所以这里简单讲解一下:
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (; ; ) { // 本场景下: node是队列里的第一个元素, 也就是await的线程对应的node. // 本场景下: p是node的前一个节点, 也就是head节点了 final Node p = node.predecessor(); // 本场景下: p==head. 锁现在空闲, tryAcquire也会成功. if (p == head && tryAcquire(arg)) { // 将node设置为新的head. head节点隐含的意思就是: head节点对应的线程是当前锁的持有者 setHead(node); p.next = null; // help GC failed = false; // 返回false. 因为本场景下该线程没有被中断过. return interrupted; } //--- 本场景下, 不会执行到下面的代码 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
最终这个方法返回了true. 接下来, await()方法继续执行剩下的几行代码就可以退出了:
这两行就是做了相应的维护操作, 和线程中断判断, 这里就不讲解了.
随后,await方法执行完了, 退出方法栈.
然后就继续往下执行. 执行System.out.println, 然后是unlock.
至此本段程序就执行完了.
原文地址:https://www.cnblogs.com/noKing/p/9367563.html