深入理解 JUC:AQS 队列同步器( 十 )


也就说对于双向链表而言 , 没有不加锁的原子手段可以保证构造双向指针的线程安全性 。 回到代码中 , 我们回顾一下往同步队列中添加结点的执行过程 , 如下(其中 pred 是末尾结点 , 而 node 是待插入的结点):
node.prev = pred;if (this.compareAndSetTail(pred, node)) {pred.next = node;return node;}上述方法会将 node 结点的 prev 指针指向 pred 结点 , 而将 pred 的 next 指针指向 node 结点的过程需要建立在基于 CAS 成功将 node 设置为末端结点的基础之上 , 如果这一过程失败则 next 指针将会断掉 , 而选择从后往前遍历则始终能够保证遍历到头结点 。
共享获取资源针对共享模式获取资源 , AbstractQueuedSynchronizer 同样定义了多个版本的 acquire 方法实现 , 包括:acquireShared、acquireSharedInterruptibly , 以及 tryAcquireSharedNanos , 其中 acquireSharedInterruptibly 是 acquireShared 的中断版本 , 在等待获取资源期间支持响应中断请求 , tryAcquireSharedNanos 除了支持响应中断以外 , 还引入了超时等待机制 。 下面同样主要分析一下 AbstractQueuedSynchronizer#acquireShared 的实现 , 理解了该方法的实现机制 , 也就自然而然理解了另外两个版本的实现机制 。
方法 AbstractQueuedSynchronizer#acquireShared 的实现如下:
public final void acquireShared(int arg) {// 返回负数表示获取资源失败if (this.tryAcquireShared(arg) < 0) {// 将当前线程添加到条件队列 , 基于自旋等待获取资源this.doAcquireShared(arg);}}private void doAcquireShared(int arg) {// 将当前线程加入条件队列末端 , 并标记为共享模式final Node node = this.addWaiter(Node.SHARED);boolean failed = true;try {boolean interrupted = false; // 标记自旋过程中是否被中断for (; ; ) {// 获取前驱结点final Node p = node.predecessor();// 如果前驱结点为头结点 , 说明当前结点是排在同步队列最前面 , 可以尝试获取资源if (p == head) {// 尝试获取资源int r = this.tryAcquireShared(arg);if (r >= 0) {// 获取资源成功 , 设置自己为头结点 , 并尝试唤醒后继结点this.setHeadAndPropagate(node, r);p.next = null; // help GCif (interrupted) {selfInterrupt();}failed = false;return;}}// 如果还未轮到当前结点 , 或者获取资源失败if (shouldParkAfterFailedAcquire(p, node) // 判断是否需要阻塞当前线程}}} finally {// 尝试获取资源失败 , 说明执行异常 , 取消当前结点获取资源的进程if (failed) {this.cancelAcquire(node);}}}上述方法与 AbstractQueuedSynchronizer#acquire 的实现逻辑大同小异 , 区别在于线程在被封装成结点之后 , 是以共享(SHARED)模式在同步队列中进行等待 。 这里我们重点关注一下 AbstractQueuedSynchronizer#setHeadAndPropagate 方法的实现 , 当结点上的线程成功获取到资源会触发执行该方法 , 以尝试唤醒后继结点 。 实现如下:
private void setHeadAndPropagate(Node node, int propagate) {Node h = head; // 记录之前的头结点this.setHead(node); // 头结点一般记录持有资源的线程结点/** 如果满足以下条件 , 尝试唤醒后继结点:** 1\. 存在剩余可用的资源;* 2\. 后继结点处于等待状态 , 或后继结点为空** Try to signal next queued node if:*Propagation was indicated by caller,*or was recorded (as h.waitStatus either before or after setHead) by a previous operation*(note: this uses sign-check of waitStatus because PROPAGATE status may transition to SIGNAL.)* and*The next node is waiting in shared mode,*or we don't know, because it appears null** The conservatism in both of these checks may cause unnecessary wake-ups,* but only when there are multiple racing acquires/releases, so most need signals now or soon anyway.*/if (propagate > 0 // 存在剩余可用的资源|| h == null || h.waitStatus