图灵学院Java架构师-VIP-并发编程（AQS详解）

标签：must   field   共享资源   void   throws   接下来   中断问题   ali   工具   

1、LockSupport

LockSupport类的核心方法其实就两个：park()和unark()，其中park()方法用来阻塞当前调用线程，unpark()方法用于唤醒指定线程
LockSupport类使用了一种名为Permit（许可）的概念来做到阻塞和唤醒线程的功能，可以把许可看成是一种(0,1)信号量（Semaphore），但与 Semaphore 不同的是，许可的累加上限是1。
初始时，permit为0，当调用unpark()方法时，线程的permit加1，当调用park()方法时，如果permit为0，则调用线程进入阻塞状态。

所以以下代码不会阻塞

// 初始信号量为0，调用unpark，信号量+1
LockSupport.unpark();
// 当前信号量为1，调用park,信号量-1
LockSupport.park();
// 以下代码可以继续执行
doSomething()

2、AQS

AbstractQueueSynchronizer是并发工具的核心，是一个抽象类，提供公平 / 非公平获取锁，获取可重入 / 不可重入锁，共享 / 排他等功能支持
AQS框架，分离了构建同步器时的一系列关注点，它的所有操作都围绕着资源——同步状态（synchronization state）来展开，并替用户解决了如下问题：

1. 资源是可以被同时访问？还是在同一时间只能被一个线程访问？（共享/独占功能）
2. 访问资源的线程如何进行并发管理？（等待队列）
3. 如果线程等不及资源了，如何从等待队列退出？（超时/中断）

这其实是一种典型的模板方法设计模式：父类（AQS框架）定义好骨架和内部操作细节，具体规则由子类去实现。

什么是资源：

AQS-API
共享和排他

1. 支持中断超时
2. 支持独占和共享
3. 支持Condition条件等待

CAS操作方法
Java中CAS操作的实现都委托给一个名为UnSafe类

等待队列

资源获取操作

资源释放：

3、等待队列

CLH队列中的结点是对线程的包装，结点一共有两种类型：独占（EXCLUSIVE）和共享（SHARED）。
每种类型的结点都有一些状态，其中独占结点使用其中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2)，共享结点使用其中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。

static final class Node {
    
    // 共享模式结点
    private static final Node SHARED = new Node();
    
    // 独占模式结点
    private?static final Node EXCLUSIVE = null;

    private?static final int CANCELLED =  1;

    private?static final int SIGNAL    = -1;

    static final int CONDITION = -2;

    static final int PROPAGATE = -3;
    // 等待状态
    volatile int waitStatus;

    // 前驱指针
    volatile Node prev;

    // 后驱指针
    volatile Node next;

    // 结点所包装的线程
    volatile Thread thread;

    // Condition队列使用，存储condition队列中的后继节点
    Node nextWaiter;

    Node() {
    }

    Node(Thread thread, Node mode) { 
        this.nextWaiter = mode;
        this.thread = thread;
    }
}

4、加锁

这里以ReentrantLock为例，看看 FairSync 和 NonfairSync 的源码

• FairSync

static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897090466540L;
    
    // 加锁
    final void lock() {
        acquire(1);
    }

    /**
     * Fair version of tryAcquire.  Don't grant access unless
     * recursive call or no waiters or is first.
     */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        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;
        }
        return false;
    }
}

public final boolean hasQueuedPredecessors() {
    // The correctness of this depends on head being initialized
    // before tail and on head.next being accurate if the current
    // thread is first in queue.
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    // 如果头和尾指向了同一个对象（null）或者头节点下一个为当前节点时，说明队列没有节点，或仅有一个当前节点
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

• NonfairSync

/**
 * Sync object for non-fair locks
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

可以发现他们两个都继承自 Sync，Sync继承自AbstractQueuedSynchronizer，在ReentrantLock中的实现如下

• Sync

abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = -5179523762034025860L;

    /**
     * Performs {@link Lock#lock}. The main reason for subclassing
     * is to allow fast path for nonfair version.
     */
    abstract void lock();

    /**
     * Performs non-fair tryLock.  tryAcquire is implemented in
     * subclasses, but both need nonfair try for trylock method.
     */
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }
}

public void lock() {
    sync.lock();
}

ReentrantLock的加锁直接委托了Sync的 lock，在Sync中，lock是个抽象方法，依次查看NonfairSync 和 FairSync 的实现，如上文源码注释

非公平锁：

1、sync 修改 state (compareAndSetState)，尝试直接获取锁，获取成功，则设置排他属性。获取失败，则执行AQS获取锁逻辑

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

2、执行获取锁，tryAcquire （NonfairSync 和 FairSync都有各自的实现），获取失败，获取等待队列，将线程放入等待队列中

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        // 将当前线程包装为独占节点加入队列
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

3、NonfairSync 对 tryAcquire 的实现

• 获取state状态
• 如果当前线程持有锁，设置nextc, 添加acquires，并设置给state，可见Reentrant支持可重入锁
• 如果当前线程没有持有锁，CAS尝试获取锁，获取成功，设置排他性，否则获取锁失败，返回false

// 委托给父类
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}
// 父类实现
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    // 获取state状态
    int c = getState();
    // 如果当前线程持有锁，设置nextc, 添加acquires，并设置给state，可见Reentrant支持可重入锁
    // 如果当前线程没有持有锁，CAS尝试获取锁，获取成功，设置排他性，否则获取锁失败，返回false
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

3、获取锁失败的处理：acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

• 为当前线程创建排队节点（这是一个双向链表）

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        // 将当前节点加入到队列尾部（算是优化吧）
        node.prev = pred;
        // 设置尾节点为当前节点
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 如果尾为null, 先初始化队列
    enq(node);
    return node;
}

// node构造函数
Node(Thread thread, Node mode) {     // Used by addWaiter
    this.nextWaiter = mode;
    this.thread = thread;
}

// 自旋将节点加入尾部，包含队列初始化
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        // Must initialize
        if (t == null) {
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

为已经在队列中的线程以独占不间断模式获取。 由条件等待方法使用以及获取

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            // 如果当前节点的前驱节点是头节点，当前线程再次获取锁，如果成功，进入if
            if (p == head && tryAcquire(arg)) {
                // 将当前节点设置为头节点，返回中断状态
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 尝试阻塞线程
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        // 退出等待
        if (failed)
            cancelAcquire(node);
    }
}

// 判断是否能阻塞当前线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    // 判断前驱节点状态
    if (ws == Node.SIGNAL)
        // SIGNAL：前驱节点释放锁时，会唤醒当前节点，可以阻塞
        return true;
    if (ws > 0) {
        // CANCELED:前驱节点已中断/取消，需要从队列中移除
        // 循环检查，剔除队列前面无效的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        // 再将当前节点放入队列
        pred.next = node;
    } else {
        // 将前驱节点修改为 SIGNAL, 自旋再次执行此方法时，将走第一条分支
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

// 阻塞当前线程
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

acquireQueued方法抛出异常时会执行 cancelAcquire

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    node.thread = null;

    // Skip cancelled predecessors
    // 再次剔除无效的前驱节点
    Node pred = node.prev;
    while (pred.waitStatus > 0)
         node.prev = pred = pred.prev;

    // 前驱节点的原后继节点，用于后续CAS操作
    Node predNext = pred.next;

    // 当前节点设置为打断
    node.waitStatus = Node.CANCELLED;

    // 如果当前节点为尾节点，设置最后一个有效节点为尾节点
    if (node == tail && compareAndSetTail(node, pred)) {
        // 有效的前驱节点属性next设置为null
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            // 唤醒后继节点
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

5、解锁

public void unlock() {
    sync.release(1);
}

public final boolean release(int arg) {
    // 尝试解锁
    if (tryRelease(arg)) {
        Node h = head;
        // 释放锁成功唤醒后继节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

释放锁，本质为维护state变量，此处支持可重入锁，如果state值为0，说明释放，取消线程的独占，并更新state

protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

唤醒后继节点

private void unparkSuccessor(Node node) {
    
    // 状态置为0，初始化
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    // 如果 s 为 null, 从后向前迭代，找到最前的未被CANCALLED的节点
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    // 唤醒后继节点
    if (s != null)
        LockSupport.unpark(s.thread);
}

后继节点被唤醒后，回到之前的逻辑，开始争夺锁，并将头节点设置为当前节点

6、其他特性

公平锁和非公平锁：

公平锁尝试加锁时，先判断队列中是否有等待线程，如果有，直接进队列

非公平锁直接获取锁，获取失败才进队列

中断特性：

如下代码中，使用了一个bool标记返回标识线程的中断状态，而中断锁会直接抛出异常

private void doAcquireInterruptibly(int arg)
            throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);//以独占模式放入队列尾部
    boolean failed = true;
    try {
        for (; ; ) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                // 中断问题，抛出异常；非中断方法中，返回中断的 bool 变量   
                throw new InterruptedException();
        }
    } finally {
        if (failed)
          cancelAcquire(node);
    }
}

限时等待：

public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout);
}

具体实现：

private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (nanosTimeout <= 0L)
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;
    final Node node = addWaiter(Node.EXCLUSIVE);// 加入队列
    boolean failed = true;
    try {
        for (; ; ) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            // 自旋时更新剩余等待时间
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                // 超时直接返回获取失败
                return false;
            if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold)
                // 阻塞指定时长，超时则线程自动被唤醒，自旋时，将在上一个if块退出
                // 底层通过 unsafe 实现
                LockSupport.parkNanos(this, nanosTimeout);
            if (Thread.interrupted())// 当前线程中断状态
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

7、条件队列

当线程在指定Condition对象上等待的时候，是将线程包装成结点，加入了条件队列，然后阻塞。当线程被通知唤醒时，则是将条件队列中的结点转换成等待队列中的结点，之后的处理就和独占功能完全一样。

J.U.C包提供了Conditon接口，用以对原生的Object.wait()、Object.notify()进行增强。

public Condition newCondition() {
        return sync.newCondition();
}
final ConditionObject newCondition() {
    return new ConditionObject();
}

在ReentrantLock中，通过内部ConditionObject实现了Condition接口，提供对条件队列的支持

public class ConditionObject implements Condition, java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    /**
         * First node of condition queue.
     */
    private transient Node firstWaiter;
    /**
     * Last node of condition queue.
     */
    private transient Node lastWaiter;
    /**
     * Creates a new {@code ConditionObject} instance.
     */
    public ConditionObject() {
    }
    ...
}

条件队列操作：

1、加入条件队列等待，条件队列入口

public final void await() throws InterruptedException {
    // 如果当前线程被中断则直接抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 把当前节点加入条件队列
    Node node = addConditionWaiter();
    // 释放掉已经获取的独占锁资源
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 如果不在同步队列中则不断挂起
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        // 这里被唤醒可能是正常的signal操作也可能是中断
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    /**
     * 走到这里说明节点已经条件满足被加入到了同步队列中或者中断了
     * 和独占锁调用同样的获取锁方法，从这里可以看出条件队列只能用于独占锁
     * 在处理中断之前首先要做的是从同步队列中成功获取锁资源
     */
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    // 走到这里说明已经成功获取到了独占锁，接下来就做些收尾工作
    // 删除条件队列中被取消的节点
    // clean up if cancelled
    if (node.nextWaiter != null) 
        unlinkCancelledWaiters();
    // 根据不同模式处理中断
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

addConditionWaiter：将当前线程包装为节点加入条件队列

/**
 * 1.与同步队列不同，条件队列头尾指针是firstWaiter跟lastWaiter
 * 2.条件队列是在获取锁之后，也就是临界区进行操作，因此很多地方不用考虑并发
 */
private Node addConditionWaiter() {
    Node t = lastWaiter;
    //如果最后一个节点被取消，则删除队列中被取消的节点
    //至于为啥是最后一个节点后面会分析
    if (t != null && t.waitStatus != Node.CONDITION) {
        //删除所有被取消的节点
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    //创建一个类型为CONDITION的节点并加入队列，由于在临界区，所以这里不用并发控制
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

删除条件队列当中被取消的节点

/**
 * 删除条件队列当中被取消的节点
 */
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            // 判断中间变量
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        } else
            // 保存的是最靠前的有效条件节点
            trail = t;
        t = next;
    }
}

释放所有资源

/**
 * 入参就是新创建的节点，即当前节点
 */
final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        //这里这个取值要注意，获取当前的state并释放，这从另一个角度说明必须是独占锁
        int savedState = getState();
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            //如果这里释放失败，则抛出异常
            throw new IllegalMonitorStateException();
        }
    } finally {
        /**
         * 如果释放锁失败，则把节点取消，由这里就能看出来上面添加节点的逻辑中
         * 只需要判断最后一个节点是否被取消就可以了
         */
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

判断节点是否在同步队列中

/**
 * 判断节点是否在同步队列中
 */
final boolean isOnSyncQueue(Node node) {
    //快速判断1：节点状态或者节点没有前置节点
    //注：同步队列是有头节点的，而条件队列没有
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    //快速判断2：next字段只有同步队列才会使用，条件队列中使用的是nextWaiter字段
    if (node.next != null) // If has successor, it must be on queue
        return true;
    //上面如果无法判断则进入复杂判断
    return findNodeFromTail(node);
}

2、唤醒等待队列

/**
 * 通知条件队列当中节点到同步队列当中去排队
 */
public final void signal() {
    // 节点不能已经持有独占锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
    /**
     * 发信号通知条件队列的节点准备到同步队列当中去排队
     */
            doSignal(first);
}

排队过程

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) && (first = firstWaiter) != null);
}

transferForSignal方法会将CONDITON结点转换为初始结点，并插入【等待队列】

final boolean transferForSignal(Node node) {
    // 尝试转化为初始节点
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
        
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

8、共享锁

AQS的共享功能，通过钩子方法tryAcquireShared暴露，与独占功能最主要的区别就是：

共享功能的结点，一旦被唤醒，会向队列后部传播（Propagate）状态，以实现共享结点的连续唤醒。这也是共享的含义，当锁被释放时，所有持有该锁的共享线程都会被唤醒，并从等待队列移除。

以CountDownLatch为例：CountDownLatch内部继承了AQS，覆盖了共享获取和释放锁的方法

构造：

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

    Sync(int count) {
        setState(count);
    }

    int getCount() {
        return getState();
    }

    protected int tryAcquireShared(int acquires) 
        // 只要 state == 0 就获取锁成功
        return (getState() == 0) ? 1 : -1;
    }

    protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        for (;;) {
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}

private final Sync sync;

await()：

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    // 响应中断
    if (Thread.interrupted())
        throw new InterruptedException();
    // 尝试获取共享锁，取决于state变量是否为0（1:获取成功；-1:获取失败）
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

doAcquireSharedInterruptibly

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
    // 以共享节点形式加入等待队列
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            // 如果前驱节点为头节点
            if (p == head) {
                // 再次获取锁
                int r = tryAcquireShared(arg);
                // 获取成功设置当前节点为头节点
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 判断阻塞条件，响应中断
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        // 移除当前节点
        if (failed)
            cancelAcquire(node);
    }
}

setHeadAndPropagate：

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; 
    setHead(node);
    // 判断头节点等待状态
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

doReleaseShared：

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                // 如果头节点等待状态为 SIGN, 设置头节点状态归0
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                // 传递唤醒后继节点
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 退出循环的条件是head并未改变
        if (h == head)                   // loop if head changed
            break;
    }
}

countDown()：

public void countDown() {
    // 释放资源
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 释放资源成功后执行
        doReleaseShared();
        return true;
    }
    return false;
}

图灵学院Java架构师-VIP-并发编程（AQS详解）

标签：must   field   共享资源   void   throws   接下来   中断问题   ali   工具   

原文地址：https://www.cnblogs.com/zuier/p/11388795.html