AbstractQueuedSynchronizer AQS源码分析

2022-06-08 13:59:07

申明：jdk版本为1.8

AbstractQueuedSynchronizer是jdk中实现锁的一个抽象类，有排他和共享两种模式。

我们这里先看排他模式，共享模式后面结合java.util.concurrent.locks.ReentrantReadWriteLock单独写一篇随笔。

后面还会分析可重入锁java.util.concurrent.locks.ReentrantLock。

言归正传，一起来看看吧。

AQS中主要是通过state变量来控制锁状态的，子类通过重写下面的某些方法并控制state值来达到获取锁或者解锁效果：

一般情况下要实现自己的锁，会实现java.util.concurrent.locks.Lock接口，并通过内部类继承自java.util.concurrent.locks.AbstractQueuedSynchronizer实现。形如下面这样：

好了，AQS的扩展就先到这里，后面讲具体的锁时再详细分析。

下面分析AQS本身的代码实现。

核心方法是java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(int)获取指定数量的锁（实际就是给state加多少的问题）以及java.util.concurrent.locks.AbstractQueuedSynchronizer.release(int)释放指定数量的锁（实际就是给state减多少的问题）

acquire

代码以及注释说明如下；

看注释就明白了，这是以排他模式获取锁，并忽略线程中断，与acquireInterruptibly相对应。acquireInterruptibly在遇到线程被中断时会抛出InterruptedException异常。

这里会先调用一次tryAcquire尝试获取锁，如果获取失败，就会新建一个Node并加到双向链表尾部，该Node会在自旋里面重复地尝试直到获得锁为止，过程中线程可能会阻塞或者不阻塞，这点完全取决于Node中的waitStatus状态值，这一点后面会详细分析。

tryAcquire是子类实现的，负责操作state值，通过state值的控制，子类可以实现各种各样的锁，父类的主要逻辑就是控制好Node的链表，线程的阻塞、唤醒等。

下面看下addWaiter方法，方法的逻辑就是用一个Node对象包装当前线程，并将Node加入到双向链表的尾部：

acquire采用的是排他模式，这里的入参是Node.EXCLUSIVE，创建一个排他模式的Node。大部分情况下tail是不为null的，看注释就知道了，

这里外部加了层判断尝试以最快方式将新建的Node挂到链表尾部。

因为是并发环境，所以compareAndSetTail有可能失败，失败的话就进入enq方法，以自旋方式往链表尾部添加。

addWaiter完之后将刚刚新建的Node传入acquireQueued，自旋尝试获得锁：

通过上面代码可知，如果node的前任node是head，并且tryAcquire获得锁成功，就将当前node设为head并返回是否是因为线程中断而从阻塞状态唤醒的。

否则，shouldParkAfterFailedAcquire，先检查是否要阻塞当前线程，如果需要阻塞，则调用parkAndCheckInterrupt，阻塞线程，

并在唤醒后检查是否是终端导致的线程唤醒，同时设置interrupted = true

下面分别看下这两个方法：

shouldParkAfterFailedAcquire：

根据前任节点的状态决定是否应该阻塞，如果前任节点状态为Node.SIGNAL就直接阻塞当前节点对应的线程，否则检查前任节点状态是否为cancelled，

如果是的话，就将前面所有状态为cancelled的节点剔除，剩下的逻辑就是前任状态设置为Node.SIGNAL，总结一句话，在自旋过程中如果没有成功获得锁

的情况下，兜兜转转最终会返回true，阻塞当前线程。

node状态描述如下：

返回true后就会执行parkAndCheckInterrupt方法，如下：

调用LockSupport.park阻塞当前线程，LockSupport的分析见我的另外一篇随笔。

这里唤醒一般通过LockSupport.unpark唤醒，注意，线程的interrupt方法也会唤醒该线程，所以这里调用了Thread.interrupted()静态方法

判断唤醒方式。唤醒后就会继续进入自旋尝试获得锁。

release

释放指定数量的锁，调用子类tryRelease，将State减去入参对应的数值，如果为0，则表示锁彻底释放，进入unparkSuccessor方法，唤醒head后面节点对应的线程。

正常情况下唤醒head后面一个节点的thread，但是，如果nextNode为cancel状态，就从tail往前找最前面符合条件的node。

至此，锁的获取、释放都讲完了，还有一个涉及到共享模式的那一路逻辑后面结合java.util.concurrent.locks.ReentrantReadWriteLock再分析。

ConditionObject

首先，一把AQS锁可以new多个ConditionObject对象，调用await和signal必须在锁的lock和unlock中间，也就是必须获得锁的情况下才能调用，否则会抛出异常。

类似synchronized中调用监视器的wait和notify，不同的是AQS同一把锁可以有多个conditionObject。

下面详细分析下AQS的await和signal。

await

/**

         * Implements interruptible condition wait.

         * <ol>

         * <li> If current thread is interrupted, throw InterruptedException.

         * <li> Save lock state returned by {@link #getState}.

         * <li> Invoke {@link #release} with saved state as argument,

         *      throwing IllegalMonitorStateException if it fails.

         * <li> Block until signalled or interrupted.

         * <li> Reacquire by invoking specialized version of

         *      {@link #acquire} with saved state as argument.

         * <li> If interrupted while blocked in step 4, throw InterruptedException.

         * </ol>

         */

        public final void await() throws InterruptedException {
　　　　　　　// 检查线程是否已经中断

            if (Thread.interrupted())

                throw new InterruptedException();
　　　　　　　// 将当前线程包装进Node，状态为Node.CONDITION，并加到条件队列末尾

            Node node = addConditionWaiter();
　　　　　　　// 释放掉该线程获取的所有锁状态

            int savedState = fullyRelease(node);

            int interruptMode = 0;
　　　　　　　// 循环等待当前node直到该node从条件队列转入等待队列
　　　　　　　// 或者该线程被中断，也就是checkInterruptWhileWaiting返回0
            while (!isOnSyncQueue(node)) {
　　　　　　　　　// 如果还没有转移到等待队列则阻塞当前线程
　　　　　　　　　// 这里的线程唤醒有一下几种情况：
　　　　　　　　　// 1.调用signal后转移到等待队列并排队到该node正常唤醒
　　　　　　　　　// 2.线程中断唤醒
　　　　　　　　　// 3.signal时调用transferForSignal入队后检查发现前驱节点取消了，或者对前驱节点设置状态为Node.SIGNAL时CAS失败
　　　　　　　　　LockSupport.park(this);
　　　　　　　　　// 唤醒后检查状态，是否被中断唤醒，如果不是返回0，如果是，再看是在signal之前还是之后，
　　　　　　　　　// 之前的话返回THROW_IE，await结束后抛出InterruptedException异常
　　　　　　　　　// 之后的话返回REINTERRUPT，需要重新设置中断标志
　　　　　　　　　// 值得说明的是，线程被中断唤醒后，不管处在条件队列的什么位置，都会直接将对应node转移到等待队列
　　　　　　　　　if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
　　　　　　　　　　　　break;
            }
　　　　　　　// 分析checkInterruptWhileWaiting可知，即使线程中断也会转移到等待队列，所以这里acquireQueued自旋排队获取锁
　　　　　　　// 如果acquireQueued返回true则说明等待过程中发生了中断，如果不是signal之前发生的中断，需要重新设置中断状态以便外部逻辑感知到
　　　　　　　// 因为checkInterruptWhileWaiting和parkAndCheckInterrupt中都是调用的Thread.interrupted静态方法，这个方法会清除中断状态
　　　　　　　if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
　　　　　　　// 正常signal的node会doSignal的时候设置nextWaiter = null，
　　　　　　　//所以这里的条件满足情况就是在直到获得锁之前都没有调用signal（因为signal开始执行就设置了nextWaiter = null）
　　　　　　　if (node.nextWaiter != null) // clean up if cancelled
　　　　　　　　　// 在上述情况下，将node从条件队列中移除
　　　　　　　　　 unlinkCancelledWaiters();
　　　　　　// 如果发生中断，则根据中断情况向外反馈
　　　　　　if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
     　　　}

整个主线逻辑就是上面分析的那样，下面详解下关键的方法：

checkInterruptWhileWaiting，检查是否是中断导致的线程唤醒并返回后续如何处理的标志：

THROW_IE：抛出异常

REINTERRUPT：重新设置中断标志

从上面代码可知，如果不是中断，则返回0，重新进入while循环判断是否在等待队列中，如果是中断，则判断中断时机：

这里的中断时机主要针对的signal之前还是之后，因为在signal中会先将node状态设置为0再将node转移到等待队列，如下图所示：

因为都是CAS操作，所以这里通过判断能否将状态由CONDITION设置为0判断是在signal之前还是之后，如果设置成功，说明在signal操作之前，则将node加入到等待队列并返回true，对应的状态是THROW_IE，如果设置失败，说明signal中的CAS操作抢先了，那就while循环等待调用signal的线程将该node转移到等待队列，对应的状态是REINTERRUPT。

码农公寓

acquire

release

ConditionObject

await

相关文章