本篇学习目标
- 回顾CLH同步队列的结构。
- 学习独占式资源获取和释放的流程。
CLH队列的结构
我在Java并发包源码学习系列：AbstractQueuedSynchronizer#同步队列与Node节点已经粗略地介绍了一下CLH的结构，本篇主要解析该同步队列的相关操作，因此在这边再回顾一下：
AQS通过内置的FIFO同步双向队列来完成资源获取线程的排队工作，内部通过节点head【实际上是虚拟节点，真正的第一个线程在head.next的位置】和tail记录队首和队尾元素，队列元素类型为Node。
- 如果当前线程获取同步状态失败（锁）时，AQS 则会将当前线程以及等待状态等信息构造成一个节点（Node）并将其加入同步队列，同时会阻塞当前线程
- 当同步状态释放时，则会把节点中的线程唤醒，使其再次尝试获取同步状态。
接下来将要通过AQS以独占式的获取和释放资源的具体案例来详解内置CLH阻塞队列的工作流程，接着往下看吧。
资源获取
1public final void acquire(int arg) { 2 if (!tryAcquire(arg) && // tryAcquire由子类实现，表示获取锁，如果成功，这个方法直接返回了 3 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 如果获取失败，执行 4 selfInterrupt(); 5 } 6
- tryAcquire(int)是AQS提供给子类实现的钩子方法，子类可以自定义实现独占式获取资源的方式，获取成功则返回true，失败则返回false。
- 如果tryAcquire方法获取资源成功就直接返回了，失败的化就会执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg))的逻辑，我们可以将其进行拆分，分为两步：addWaiter(Node.EXCLUSIVE)：将该线程包装成为独占式的节点，加入队列中。acquireQueued(node,arg)：如果当前节点是等待节点的第一个，即head.next，就尝试获取资源。如果该方法返回true，则会进入selfInterrupt()的逻辑，进行阻塞。
接下来我们分别来看看addWaiter和acquireQueued两个方法。
入队Node addWaiter(Node mode)
根据传入的mode参数决定独占或共享模式，为当前线程创建节点，并入队。
1// 其实就是把当前线程包装一下，设置模式，形成节点，加入队列 2 private Node addWaiter(Node mode) { 3 // 根据mode和thread创建节点 4 Node node = new Node(Thread.currentThread(), mode); 5 // 记录一下原尾节点 6 Node pred = tail; 7 // 尾节点不为null，队列不为空，快速尝试加入队尾。 8 if (pred != null) { 9 // 让node的prev指向尾节点 10 node.prev = pred; 11 // CAS操作设置node为新的尾节点，tail = node 12 if (compareAndSetTail(pred, node)) { 13 // 设置成功，让原尾节点的next指向新的node，实现双向链接 14 pred.next = node; 15 // 入队成功，返回 16 return node; 17 } 18 } 19 // 快速入队失败，进行不断尝试 20 enq(node); 21 return node; 22 }
几个注意点：
- 入队的操作其实就是将线程通过指定模式包装为Node节点，如果队列尾节点不为null，利用CAS尝试快速加入队尾。
快速入队失败的原因有两个：
- 队列为空，即还没有进行初始化。
- CAS设置尾节点的时候失败。
- 在第一次快速入队失败后，将会走到enq(node)逻辑，不断进行尝试，直到设置成功。
不断尝试Node enq(final Node node)
1private Node enq(final Node node) { 2 // 自旋，俗称死循环，直到设置成功为止 3 for (;;) { 4 // 记录原尾节点 5 Node t = tail; 6 // 第一种情况：队列为空，原先head和tail都为null， 7 // 通过CAS设置head为哨兵节点，如果设置成功，tail也指向哨兵节点 8 if (t == null) { // Must initialize 9 // 初始化head节点 10 if (compareAndSetHead(new Node())) 11 // tail指向head，下个线程来的时候，tail就不为null了，就走到了else分支 12 tail = head; 13 // 第二种情况：CAS设置尾节点失败的情况，和addWaiter一样，只不过它在for(;;)中 14 } else { 15 // 入队，将新节点的prev指向tail 16 node.prev = t; 17 // CAS设置node为尾部节点 18 if (compareAndSetTail(t, node)) { 19 //原来的tail的next指向node 20 t.next = node; 21 return t; 22 } 23 } 24 } 25 } 26
enq的过程是自选设置队尾的过程，如果设置成功，就返回。如果设置失败，则一直尝试设置，理念就是，我总能等待设置成功那一天。
我们还可以发现，head是延迟初始化的，在第一个节点尝试入队的时候，head为null，这时使用了new Node()创建了一个不代表任何线程的节点，作为虚拟头节点，且我们需要注意它的waitStatus初始化为0，这一点对我们之后分析有指导意义。
如果是CAS失败导致重复尝试，那就还是让他继续CAS好了。
boolean acquireQueued(Node, int)
1// 这个方法如果返回true，代码将进入selfInterrupt() 2 final boolean acquireQueued(final Node node, int arg) { 3 // 注意默认为true 4 boolean failed = true; 5 try { 6 // 是否中断 7 boolean interrupted = false; 8 // 自旋，即死循环 9 for (;;) { 10 // 得到node的前驱节点 11 final Node p = node.predecessor(); 12 // 我们知道head是虚拟的头节点，p==head表示如果node为阻塞队列的第一个真实节点 13 // 就执行tryAcquire逻辑，这里tryAcquire也需要由子类实现 14 if (p == head && tryAcquire(arg)) { 15 // tryAcquire获取成功走到这，执行setHead出队操作 16 setHead(node); 17 p.next = null; // help GC 18 failed = false; 19 return interrupted; 20 } 21 // 走到这有两种情况 1.node不是第一个节点 2.tryAcquire争夺锁失败了 22 // 这里就判断 如果当前线程争锁失败，是否需要挂起当前这个线程 23 if (shouldParkAfterFailedAcquire(p, node) && 24 parkAndCheckInterrupt()) 25 interrupted = true; 26 } 27 } finally { 28 // 死循环退出，只有tryAcquire获取锁失败的时候failed才为true 29 if (failed) 30 cancelAcquire(node); 31 } 32 } 33
出队void setHead(Node)
CLU同步队列遵循FIFO，首节点的线程释放同步状态后，唤醒下一个节点。将队首节点出队的操作实际上就是，将head指针指向将要出队的节点就可以了。
1private void setHead(Node node) { 2 // head指针指向node 3 head = node; 4 // 释放资源 5 node.thread = null; 6 node.prev = null; 7 }
boolean shouldParkAfterFailedAcquire(Node,Node)
1/** 2 * 走到这有两种情况 1.node不是第一个节点 2.tryAcquire争夺锁失败了 3 * 这里就判断 如果当前线程争锁失败，是否需要挂起当前这个线程 4 * 5 * 这里pred是前驱节点， node就是当前节点 6 */ 7 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 8 // 前驱节点的waitStatus 9 int ws = pred.waitStatus; 10 // 前驱节点为SIGNAL【-1】直接返回true，表示当前节点可以被直接挂起 11 if (ws == Node.SIGNAL) 12 return true; 13 // ws>0 CANCEL 说明前驱节点取消了排队 14 if (ws > 0) { 15 // 下面这段循环其实就是跳过所有取消的节点，找到第一个正常的节点 16 do { 17 node.prev = pred = pred.prev; 18 } while (pred.waitStatus > 0); 19 // 将该节点的后继指向node，建立双向连接 20 pred.next = node; 21 } else { 22 /* 23 * waitStatus must be 0 or PROPAGATE. Indicate that we 24 * need a signal, but don't park yet. Caller will need to 25 * retry to make sure it cannot acquire before parking. 26 * 官方说明：走到这waitStatus只能是0或propagate，默认情况下，当有新节点入队时，waitStatus总是为0 27 * 下面用CAS操作将前驱节点的waitStatus值设置为signal 28 */ 29 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); 30 } 31 // 返回false，接着会再进入循环，此时前驱节点为signal，返回true 32 return false; 33 }
针对前驱节点的waitStatus有三种情况：
等待状态不会为 Node.CONDITION ，因为它用在 ConditonObject 中
- ws==-1，即为Node.SIGNAL,表示当前节点node可以被直接挂起，在pred线程释放同步状态时，会对node线程进行唤醒。
- ws > 0，即为Node.CANCELLED，说明前驱节点已经取消了排队【可能是超时，可能是被中断】，则需要找到前面没有取消的前驱节点，一直找，直到找到为止。
- ws == 0 or ws == Node.PROPAGATE：
- 默认情况下，当有新节点入队时，waitStatus总是为0，用CAS操作将前驱节点的waitStatus值设置为signal，下一次进来的时候，就走到了第一个分支。
- 当释放锁的时候，会将占用锁的节点的ws状态更新为0。
PROPAGATE表示共享模式下，前驱节点不仅会唤醒后继节点，同时也可能会唤醒后继的后继。
我们可以发现，这个方法在第一次走进来的时候是不会返回true的。原因在于，返回true的条件时前驱节点的状态为SIGNAL，而第一次的时候还没有给前驱节点设置SIGNAL呢，只有在CAS设置了状态之后，第二次进来才会返回true。
那SIGNAL的意义到底是什么呢？
这里引用：并发编程——详解 AQS CLH 锁 # 为什么 AQS 需要一个虚拟 head 节点 waitStatus这里用ws简称，每个节点都有ws变量，用于表示该节点的状态。初始化的时候为0，如果被取消为1，signal为-1。 如果某个节点的状态是signal的，那么在该节点释放锁的时候，它需要唤醒下一个节点。 因此，每个节点在休眠之前，如果没有将前驱节点的ws设置为signal，那么它将永远无法被唤醒。 因此我们会发现上面当前驱节点的ws为0或propagate的时候，采用cas操作将ws设置为signal，目的就是让上一个节点释放锁的时候能够通知自己。
boolean parkAndCheckInterrupt()
1private final boolean parkAndCheckInterrupt() { 2 // 挂起当前线程 3 LockSupport.park(this); 4 return Thread.interrupted(); 5 }
shouldParkAfterFailedAcquire方法返回true之后，就会调用该方法，挂起当前线程。
LockSupport.park(this)方法挂起的线程有两种途径被唤醒：1.被unpark() 2.被interrupt()。
需要注意这里的Thread.interrupted()会清除中断标记位。
void cancelAcquire(node)
上面tryAcquire获取锁失败的时候，会走到这个方法。
1private void cancelAcquire(Node node) { 2 // Ignore if node doesn't exist 3 if (node == null) 4 return; 5 // 将节点的线程置空 6 node.thread = null; 7 8 // 跳过所有的取消的节点 9 Node pred = node.prev; 10 while (pred.waitStatus > 0) 11 node.prev = pred = pred.prev; 12 13 // predNext is the apparent node to unsplice. CASes below will 14 // fail if not, in which case, we lost race vs another cancel 15 // or signal, so no further action is necessary. 16 // 这里在没有并发的情况下，preNext和node是一致的 17 Node predNext = pred.next; 18 19 // Can use unconditional write instead of CAS here. 可以直接写而不是用CAS 20 // After this atomic step, other Nodes can skip past us. 21 // Before, we are free of interference from other threads. 22 // 设置node节点为取消状态 23 node.waitStatus = Node.CANCELLED; 24 25 // 如果node为尾节点就CAS将pred设置为新尾节点 26 if (node == tail && compareAndSetTail(node, pred)) { 27 // 设置成功之后，CAS将pred的下一个节点置为空 28 compareAndSetNext(pred, predNext, null); 29 } else { 30 // If successor needs signal, try to set pred's next-link 31 // so it will get one. Otherwise wake it up to propagate. 32 int ws; 33 if (pred != head && // pred不是首节点 34 ((ws = pred.waitStatus) == Node.SIGNAL || // pred的ws为SIGNAL 或 可以被CAS设置为SIGNAL 35 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && 36 pred.thread != null) { // pred线程非空 37 // 保存node 的下一个节点 38 Node next = node.next; 39 // node的下一个节点不是cancelled，就cas设置pred的下一个节点为next 40 if (next != null && next.waitStatus <= 0) 41 compareAndSetNext(pred, predNext, next); 42 } else { 43 // 上面的情况除外，则走到这个分支，唤醒node的下一个可唤醒节点线程 44 unparkSuccessor(node); 45 } 46 47 node.next = node; // help GC 48 } 49 }
释放资源
boolean release(int arg)
1public final boolean release(int arg) { 2 if (tryRelease(arg)) { // 子类实现tryRelease方法 3 // 获得当前head 4 Node h = head; 5 // head不为null并且head的等待状态不为0 6 if (h != null && h.waitStatus != 0) 7 // 唤醒下一个可以被唤醒的线程，不一定是next哦 8 unparkSuccessor(h); 9 return true; 10 } 11 return false; 12 } 13
- tryRelease(int)是AQS提供给子类实现的钩子方法，子类可以自定义实现独占式释放资源的方式，释放成功并返回true，否则返回false。
- unparkSuccessor(node)方法用于唤醒等待队列中下一个可以被唤醒的线程，不一定是下一个节点next，比如它可能是取消状态。
- head 的ws必须不等于0，为什么呢？当一个节点尝试挂起自己之前，都会将前置节点设置成SIGNAL -1，就算是第一个加入队列的节点，在获取锁失败后，也会将虚拟节点设置的 ws 设置成 SIGNAL，而这个判断也是防止多线程重复释放，接下来我们也能看到释放的时候，将ws设置为0的操作。
void unparkSuccessor(Node node)
1private void unparkSuccessor(Node node) { 2 /* 3 * If status is negative (i.e., possibly needing signal) try 4 * to clear in anticipation of signalling. It is OK if this 5 * fails or if status is changed by waiting thread. 6 */ 7 int ws = node.waitStatus; 8 // 如果node的waitStatus<0为signal，CAS修改为0 9 // 将 head 节点的 ws 改成 0，清除信号。表示，他已经释放过了。不能重复释放。 10 if (ws < 0) 11 compareAndSetWaitStatus(node, ws, 0); 12 13 /* 14 * Thread to unpark is held in successor, which is normally 15 * just the next node. But if cancelled or apparently null, 16 * traverse backwards from tail to find the actual 17 * non-cancelled successor. 18 */ 19 // 唤醒后继节点，但是有可能后继节点取消了等待 即 waitStatus == 1 20 Node s = node.next; 21 // 如果后继节点为空或者它已经放弃锁了 22 if (s == null || s.waitStatus > 0) { 23 s = null; 24 // 从队尾往前找，找到没有没取消的所有节点排在最前面的【直到t为null或t==node才退出循环嘛】 25 for (Node t = tail; t != null && t != node; t = t.prev) 26 // 如果>0表示节点被取消了，就一直向前找呗，找到之后不会return，还会一直向前 27 if (t.waitStatus <= 0) 28 s = t; 29 } 30 // 如果后继节点存在且没有被取消，会走到这，直接唤醒后继节点即可 31 if (s != null) 32 LockSupport.unpark(s.thread); 33 }
如果觉得本文对你有帮助，可以关注一下我公众号，回复关键字【面试】即可得到一份Java核心知识点整理与一份面试大礼包！另有更多技术干货文章以及相关资料共享，大家一起学习进步！
