【JUC】JDK1.8源码分析之AbstractQueuedSynchronizer-白红宇

【JUC】JDK1.8源码分析之AbstractQueuedSynchronizer

阅读量：6046 次

发布时间：2019-06-20

本文共 45046 字，大约阅读时间需要 150 分钟。

一、前言

　　在锁框架中，AbstractQueuedSynchronizer抽象类可以毫不夸张的说，占据着核心地位，它提供了一个基于FIFO队列，可以用于构建锁或者其他相关同步装置的基础框架。所以很有必要好好分析。

二、AbstractQueuedSynchronizer数据结构

　　分析类，首先就要分析底层采用了何种数据结构，抓住核心点进行分析，经过分析可知，AbstractQueuedSynchronizer类的数据结构如下

　　说明：AbstractQueuedSynchronizer类底层的数据结构是使用双向链表，是队列的一种实现，故也可看成是队列，其中Sync queue，即同步队列，是双向链表，包括head结点和tail结点，head结点主要用作后续的调度。而Condition queue不是必须的，其是一个单向链表，只有当使用Condition时，才会存在此单向链表。并且可能会有多个Condition queue。

三、AbstractQueuedSynchronizer源码分析

3.1 类的继承关系

public abstract class AbstractQueuedSynchronizer    extends AbstractOwnableSynchronizer    implements java.io.Serializable

说明：从类继承关系可知，AbstractQueuedSynchronizer继承自AbstractOwnableSynchronizer抽象类，并且实现了Serializable接口，可以进行序列化。而AbstractOwnableSynchronizer抽象类的源码如下

1 public abstract class AbstractOwnableSynchronizer 2     implements java.io.Serializable { 3      4     // 版本序列号 5     private static final long serialVersionUID = 3737899427754241961L; 6     // 构造函数 7     protected AbstractOwnableSynchronizer() { } 8     // 独占模式下的线程 9     private transient Thread exclusiveOwnerThread;10     11     // 设置独占线程 12     protected final void setExclusiveOwnerThread(Thread thread) {13         exclusiveOwnerThread = thread;14     }15     16     // 获取独占线程 17     protected final Thread getExclusiveOwnerThread() {18         return exclusiveOwnerThread;19     }20 }

说明：AbstractOwnableSynchronizer抽象类中，可以设置独占资源线程和获取独占资源线程。分别为setExclusiveOwnerThread与getExclusiveOwnerThread方法，这两个方法会被子类调用。

3.2 类的内部类

AbstractQueuedSynchronizer类有两个内部类，分别为Node类与ConditionObject类。下面分别做介绍。

1. Node类

1 static final class Node { 2         // 模式，分为共享与独占 3         // 共享模式 4         static final Node SHARED = new Node(); 5         // 独占模式 6         static final Node EXCLUSIVE = null;         7         // 结点状态 8         // CANCELLED，值为1，表示当前的线程被取消 9         // SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark10         // CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中11         // PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行12         // 值为0，表示当前节点在sync队列中，等待着获取锁13         static final int CANCELLED =  1;14         static final int SIGNAL    = -1;15         static final int CONDITION = -2;16         static final int PROPAGATE = -3;        17 18         // 结点状态19         volatile int waitStatus;        20         // 前驱结点21         volatile Node prev;    22         // 后继结点23         volatile Node next;        24         // 结点所对应的线程25         volatile Thread thread;        26         // 下一个等待者27         Node nextWaiter;28         29         // 结点是否在共享模式下等待30         final boolean isShared() {31             return nextWaiter == SHARED;32         }33         34         // 获取前驱结点，若前驱结点为空，抛出异常35         final Node predecessor() throws NullPointerException {36             // 保存前驱结点37             Node p = prev; 38             if (p == null) // 前驱结点为空，抛出异常39                 throw new NullPointerException();40             else // 前驱结点不为空，返回41                 return p;42         }43         44         // 无参构造函数45         Node() {    // Used to establish initial head or SHARED marker46         }47         48         // 构造函数49          Node(Thread thread, Node mode) {    // Used by addWaiter50             this.nextWaiter = mode;51             this.thread = thread;52         }53         54         // 构造函数55         Node(Thread thread, int waitStatus) { // Used by Condition56             this.waitStatus = waitStatus;57             this.thread = thread;58         }59 }

说明：每个线程被阻塞的线程都会被封装成一个Node结点，放入队列。每个节点包含了一个Thread类型的引用，并且每个节点都存在一个状态，具体状态如下。

　　① CANCELLED，值为1，表示当前的线程被取消。

　　② SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，需要进行unpark操作。

　　③ CONDITION，值为-2，表示当前节点在等待condition，也就是在condition queue中。

　　④ PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行。

　　⑤ 值为0，表示当前节点在sync queue中，等待着获取锁。

2. ConditionObject类

1 // 内部类  2     public class ConditionObject implements Condition, java.io.Serializable {  3         // 版本号  4         private static final long serialVersionUID = 1173984872572414699L;  5         /** First node of condition queue. */  6         // condition队列的头结点  7         private transient Node firstWaiter;  8         /** Last node of condition queue. */  9         // condition队列的尾结点 10         private transient Node lastWaiter; 11  12         /** 13          * Creates a new {
     @code ConditionObject} instance. 14          */ 15         // 构造函数 16         public ConditionObject() { } 17  18         // Internal methods 19  20         /** 21          * Adds a new waiter to wait queue. 22          * @return its new wait node 23          */ 24         // 添加新的waiter到wait队列 25         private Node addConditionWaiter() { 26             // 保存尾结点 27             Node t = lastWaiter; 28             // If lastWaiter is cancelled, clean out. 29             if (t != null && t.waitStatus != Node.CONDITION) { // 尾结点不为空，并且尾结点的状态不为CONDITION 30                 // 清除状态为CONDITION的结点 31                 unlinkCancelledWaiters();  32                 // 将最后一个结点重新赋值给t 33                 t = lastWaiter; 34             } 35             // 新建一个结点 36             Node node = new Node(Thread.currentThread(), Node.CONDITION); 37             if (t == null) // 尾结点为空 38                 // 设置condition队列的头结点 39                 firstWaiter = node; 40             else // 尾结点不为空 41                 // 设置为节点的nextWaiter域为node结点 42                 t.nextWaiter = node; 43             // 更新condition队列的尾结点 44             lastWaiter = node; 45             return node; 46         } 47  48         /** 49          * Removes and transfers nodes until hit non-cancelled one or 50          * null. Split out from signal in part to encourage compilers 51          * to inline the case of no waiters. 52          * @param first (non-null) the first node on condition queue 53          */ 54         private void doSignal(Node first) { 55             // 循环 56             do { 57                 if ( (firstWaiter = first.nextWaiter) == null) // 该节点的nextWaiter为空 58                     // 设置尾结点为空 59                     lastWaiter = null; 60                 // 设置first结点的nextWaiter域 61                 first.nextWaiter = null; 62             } while (!transferForSignal(first) && 63                      (first = firstWaiter) != null); // 将结点从condition队列转移到sync队列失败并且condition队列中的头结点不为空，一直循环 64         } 65  66         /** 67          * Removes and transfers all nodes. 68          * @param first (non-null) the first node on condition queue 69          */ 70         private void doSignalAll(Node first) { 71             // condition队列的头结点尾结点都设置为空 72             lastWaiter = firstWaiter = null; 73             // 循环 74             do { 75                 // 获取first结点的nextWaiter域结点 76                 Node next = first.nextWaiter; 77                 // 设置first结点的nextWaiter域为空 78                 first.nextWaiter = null; 79                 // 将first结点从condition队列转移到sync队列 80                 transferForSignal(first); 81                 // 重新设置first 82                 first = next; 83             } while (first != null); 84         } 85  86         /** 87          * Unlinks cancelled waiter nodes from condition queue. 88          * Called only while holding lock. This is called when 89          * cancellation occurred during condition wait, and upon 90          * insertion of a new waiter when lastWaiter is seen to have 91          * been cancelled. This method is needed to avoid garbage 92          * retention in the absence of signals. So even though it may 93          * require a full traversal, it comes into play only when 94          * timeouts or cancellations occur in the absence of 95          * signals. It traverses all nodes rather than stopping at a 96          * particular target to unlink all pointers to garbage nodes 97          * without requiring many re-traversals during cancellation 98          * storms. 99          */100         // 从condition队列中清除状态为CANCEL的结点101         private void unlinkCancelledWaiters() {102             // 保存condition队列头结点103             Node t = firstWaiter;104             Node trail = null;105             while (t != null) { // t不为空106                 // 下一个结点107                 Node next = t.nextWaiter;108                 if (t.waitStatus != Node.CONDITION) { // t结点的状态不为CONDTION状态109                     // 设置t节点的额nextWaiter域为空110                     t.nextWaiter = null;111                     if (trail == null) // trail为空112                         // 重新设置condition队列的头结点113                         firstWaiter = next;114                     else // trail不为空115                         // 设置trail结点的nextWaiter域为next结点116                         trail.nextWaiter = next;117                     if (next == null) // next结点为空118                         // 设置condition队列的尾结点119                         lastWaiter = trail;120                 }121                 else // t结点的状态为CONDTION状态122                     // 设置trail结点123                     trail = t;124                 // 设置t结点125                 t = next;126             }127         }128 129         // public methods130 131         /**132          * Moves the longest-waiting thread, if one exists, from the133          * wait queue for this condition to the wait queue for the134          * owning lock.135          *136          * @throws IllegalMonitorStateException if {
     @link #isHeldExclusively}137          *         returns {
     @code false}138          */139         // 唤醒一个等待线程。如果所有的线程都在等待此条件，则选择其中的一个唤醒。在从 await 返回之前，该线程必须重新获取锁。140         public final void signal() {141             if (!isHeldExclusively()) // 不被当前线程独占，抛出异常142                 throw new IllegalMonitorStateException();143             // 保存condition队列头结点144             Node first = firstWaiter;145             if (first != null) // 头结点不为空146                 // 唤醒一个等待线程147                 doSignal(first);148         }149 150         /**151          * Moves all threads from the wait queue for this condition to152          * the wait queue for the owning lock.153          *154          * @throws IllegalMonitorStateException if {
     @link #isHeldExclusively}155          *         returns {
     @code false}156          */157         // 唤醒所有等待线程。如果所有的线程都在等待此条件，则唤醒所有线程。在从 await 返回之前，每个线程都必须重新获取锁。158         public final void signalAll() {159             if (!isHeldExclusively()) // 不被当前线程独占，抛出异常160                 throw new IllegalMonitorStateException();161             // 保存condition队列头结点162             Node first = firstWaiter;163             if (first != null) // 头结点不为空164                 // 唤醒所有等待线程165                 doSignalAll(first);166         }167 168         /**169          * Implements uninterruptible condition wait.170          * 
      
       171          * 
        Save lock state returned by {
     @link #getState}.172          * 
        Invoke {
     @link #release} with saved state as argument,173          *      throwing IllegalMonitorStateException if it fails.174          * 
        Block until signalled.175          * 
        Reacquire by invoking specialized version of176          *      {
     @link #acquire} with saved state as argument.177          * 
      
178          */179         // 等待，当前线程在接到信号之前一直处于等待状态，不响应中断180         public final void awaitUninterruptibly() {181             // 添加一个结点到等待队列182             Node node = addConditionWaiter();183             // 获取释放的状态184             int savedState = fullyRelease(node);185             boolean interrupted = false;186             while (!isOnSyncQueue(node)) { // 187                 // 阻塞当前线程188                 LockSupport.park(this);189                 if (Thread.interrupted()) // 当前线程被中断190                     // 设置interrupted状态191                     interrupted = true; 192             }193             if (acquireQueued(node, savedState) || interrupted) // 194                 selfInterrupt();195         }196 197         /*198          * For interruptible waits, we need to track whether to throw199          * InterruptedException, if interrupted while blocked on200          * condition, versus reinterrupt current thread, if201          * interrupted while blocked waiting to re-acquire.202          */203 204         /** Mode meaning to reinterrupt on exit from wait */205         private static final int REINTERRUPT =  1;206         /** Mode meaning to throw InterruptedException on exit from wait */207         private static final int THROW_IE    = -1;208 209         /**210          * Checks for interrupt, returning THROW_IE if interrupted211          * before signalled, REINTERRUPT if after signalled, or212          * 0 if not interrupted.213          */214         private int checkInterruptWhileWaiting(Node node) {215             return Thread.interrupted() ?216                 (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :217                 0; 218         }219 220         /**221          * Throws InterruptedException, reinterrupts current thread, or222          * does nothing, depending on mode.223          */224         private void reportInterruptAfterWait(int interruptMode)225             throws InterruptedException {226             if (interruptMode == THROW_IE)227                 throw new InterruptedException();228             else if (interruptMode == REINTERRUPT)229                 selfInterrupt();230         }231 232         /**233          * Implements interruptible condition wait.234          * 
      
       235          * 
        If current thread is interrupted, throw InterruptedException.236          * 
        Save lock state returned by {
     @link #getState}.237          * 
        Invoke {
     @link #release} with saved state as argument,238          *      throwing IllegalMonitorStateException if it fails.239          * 
        Block until signalled or interrupted.240          * 
        Reacquire by invoking specialized version of241          *      {
     @link #acquire} with saved state as argument.242          * 
        If interrupted while blocked in step 4, throw InterruptedException.243          * 
      
244          */245         // // 等待，当前线程在接到信号或被中断之前一直处于等待状态246         public final void await() throws InterruptedException {247             if (Thread.interrupted()) // 当前线程被中断，抛出异常248                 throw new InterruptedException();249             // 在wait队列上添加一个结点250             Node node = addConditionWaiter();251             // 252             int savedState = fullyRelease(node);253             int interruptMode = 0;254             while (!isOnSyncQueue(node)) {255                 // 阻塞当前线程256                 LockSupport.park(this);257                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) // 检查结点等待时的中断类型258                     break;259             }260             if (acquireQueued(node, savedState) && interruptMode != THROW_IE)261                 interruptMode = REINTERRUPT;262             if (node.nextWaiter != null) // clean up if cancelled263                 unlinkCancelledWaiters();264             if (interruptMode != 0)265                 reportInterruptAfterWait(interruptMode);266         }267 268         /**269          * Implements timed condition wait.270          * 
      
       271          * 
        If current thread is interrupted, throw InterruptedException.272          * 
        Save lock state returned by {
     @link #getState}.273          * 
        Invoke {
     @link #release} with saved state as argument,274          *      throwing IllegalMonitorStateException if it fails.275          * 
        Block until signalled, interrupted, or timed out.276          * 
        Reacquire by invoking specialized version of277          *      {
     @link #acquire} with saved state as argument.278          * 
        If interrupted while blocked in step 4, throw InterruptedException.279          * 
      
280          */281         // 等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态 282         public final long awaitNanos(long nanosTimeout)283                 throws InterruptedException {284             if (Thread.interrupted())285                 throw new InterruptedException();286             Node node = addConditionWaiter();287             int savedState = fullyRelease(node);288             final long deadline = System.nanoTime() + nanosTimeout;289             int interruptMode = 0;290             while (!isOnSyncQueue(node)) {291                 if (nanosTimeout <= 0L) {292                     transferAfterCancelledWait(node);293                     break;294                 }295                 if (nanosTimeout >= spinForTimeoutThreshold)296                     LockSupport.parkNanos(this, nanosTimeout);297                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)298                     break;299                 nanosTimeout = deadline - System.nanoTime();300             }301             if (acquireQueued(node, savedState) && interruptMode != THROW_IE)302                 interruptMode = REINTERRUPT;303             if (node.nextWaiter != null)304                 unlinkCancelledWaiters();305             if (interruptMode != 0)306                 reportInterruptAfterWait(interruptMode);307             return deadline - System.nanoTime();308         }309 310         /**311          * Implements absolute timed condition wait.312          * 
      
       313          * 
        If current thread is interrupted, throw InterruptedException.314          * 
        Save lock state returned by {
     @link #getState}.315          * 
        Invoke {
     @link #release} with saved state as argument,316          *      throwing IllegalMonitorStateException if it fails.317          * 
        Block until signalled, interrupted, or timed out.318          * 
        Reacquire by invoking specialized version of319          *      {
     @link #acquire} with saved state as argument.320          * 
        If interrupted while blocked in step 4, throw InterruptedException.321          * 
        If timed out while blocked in step 4, return false, else true.322          * 
      
323          */324         // 等待，当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态325         public final boolean awaitUntil(Date deadline)326                 throws InterruptedException {327             long abstime = deadline.getTime();328             if (Thread.interrupted())329                 throw new InterruptedException();330             Node node = addConditionWaiter();331             int savedState = fullyRelease(node);332             boolean timedout = false;333             int interruptMode = 0;334             while (!isOnSyncQueue(node)) {335                 if (System.currentTimeMillis() > abstime) {336                     timedout = transferAfterCancelledWait(node);337                     break;338                 }339                 LockSupport.parkUntil(this, abstime);340                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)341                     break;342             }343             if (acquireQueued(node, savedState) && interruptMode != THROW_IE)344                 interruptMode = REINTERRUPT;345             if (node.nextWaiter != null)346                 unlinkCancelledWaiters();347             if (interruptMode != 0)348                 reportInterruptAfterWait(interruptMode);349             return !timedout;350         }351 352         /**353          * Implements timed condition wait.354          * 
      
       355          * 
        If current thread is interrupted, throw InterruptedException.356          * 
        Save lock state returned by {
     @link #getState}.357          * 
        Invoke {
     @link #release} with saved state as argument,358          *      throwing IllegalMonitorStateException if it fails.359          * 
        Block until signalled, interrupted, or timed out.360          * 
        Reacquire by invoking specialized version of361          *      {
     @link #acquire} with saved state as argument.362          * 
        If interrupted while blocked in step 4, throw InterruptedException.363          * 
        If timed out while blocked in step 4, return false, else true.364          * 
      
365          */366         // 等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。此方法在行为上等效于：awaitNanos(unit.toNanos(time)) > 0367         public final boolean await(long time, TimeUnit unit)368                 throws InterruptedException {369             long nanosTimeout = unit.toNanos(time);370             if (Thread.interrupted())371                 throw new InterruptedException();372             Node node = addConditionWaiter();373             int savedState = fullyRelease(node);374             final long deadline = System.nanoTime() + nanosTimeout;375             boolean timedout = false;376             int interruptMode = 0;377             while (!isOnSyncQueue(node)) {378                 if (nanosTimeout <= 0L) {379                     timedout = transferAfterCancelledWait(node);380                     break;381                 }382                 if (nanosTimeout >= spinForTimeoutThreshold)383                     LockSupport.parkNanos(this, nanosTimeout);384                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)385                     break;386                 nanosTimeout = deadline - System.nanoTime();387             }388             if (acquireQueued(node, savedState) && interruptMode != THROW_IE)389                 interruptMode = REINTERRUPT;390             if (node.nextWaiter != null)391                 unlinkCancelledWaiters();392             if (interruptMode != 0)393                 reportInterruptAfterWait(interruptMode);394             return !timedout;395         }396 397         //  support for instrumentation398 399         /**400          * Returns true if this condition was created by the given401          * synchronization object.402          *403          * @return {
     @code true} if owned404          */405         final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {406             return sync == AbstractQueuedSynchronizer.this;407         }408 409         /**410          * Queries whether any threads are waiting on this condition.411          * Implements {
     @link AbstractQueuedSynchronizer#hasWaiters(ConditionObject)}.412          *413          * @return {
     @code true} if there are any waiting threads414          * @throws IllegalMonitorStateException if {
     @link #isHeldExclusively}415          *         returns {
     @code false}416          */417         //  查询是否有正在等待此条件的任何线程418         protected final boolean hasWaiters() {419             if (!isHeldExclusively())420                 throw new IllegalMonitorStateException();421             for (Node w = firstWaiter; w != null; w = w.nextWaiter) {422                 if (w.waitStatus == Node.CONDITION)423                     return true;424             }425             return false;426         }427 428         /**429          * Returns an estimate of the number of threads waiting on430          * this condition.431          * Implements {
     @link AbstractQueuedSynchronizer#getWaitQueueLength(ConditionObject)}.432          *433          * @return the estimated number of waiting threads434          * @throws IllegalMonitorStateException if {
     @link #isHeldExclusively}435          *         returns {
     @code false}436          */437         // 返回正在等待此条件的线程数估计值438         protected final int getWaitQueueLength() {439             if (!isHeldExclusively())440                 throw new IllegalMonitorStateException();441             int n = 0;442             for (Node w = firstWaiter; w != null; w = w.nextWaiter) {443                 if (w.waitStatus == Node.CONDITION)444                     ++n;445             }446             return n;447         }448 449         /**450          * Returns a collection containing those threads that may be451          * waiting on this Condition.452          * Implements {
     @link AbstractQueuedSynchronizer#getWaitingThreads(ConditionObject)}.453          *454          * @return the collection of threads455          * @throws IllegalMonitorStateException if {
     @link #isHeldExclusively}456          *         returns {
     @code false}457          */458         // 返回包含那些可能正在等待此条件的线程集合459         protected final Collection
      
        getWaitingThreads() {460             if (!isHeldExclusively())461                 throw new IllegalMonitorStateException();462             ArrayList
       
         list = new ArrayList
        
         ();463             for (Node w = firstWaiter; w != null; w = w.nextWaiter) {464                 if (w.waitStatus == Node.CONDITION) {465                     Thread t = w.thread;466                     if (t != null)467                         list.add(t);468                 }469             }470             return list;471         }472 }

View Code

说明：此类实现了Condition接口，Condition接口定义了条件操作规范，具体如下

1 public interface Condition { 2  3     // 等待，当前线程在接到信号或被中断之前一直处于等待状态 4     void await() throws InterruptedException; 5      6     // 等待，当前线程在接到信号之前一直处于等待状态，不响应中断 7     void awaitUninterruptibly(); 8      9     //等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态 10     long awaitNanos(long nanosTimeout) throws InterruptedException;11     12     // 等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。此方法在行为上等效于：awaitNanos(unit.toNanos(time)) > 013     boolean await(long time, TimeUnit unit) throws InterruptedException;14     15     // 等待，当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态16     boolean awaitUntil(Date deadline) throws InterruptedException;17     18     // 唤醒一个等待线程。如果所有的线程都在等待此条件，则选择其中的一个唤醒。在从 await 返回之前，该线程必须重新获取锁。19     void signal();20     21     // 唤醒所有等待线程。如果所有的线程都在等待此条件，则唤醒所有线程。在从 await 返回之前，每个线程都必须重新获取锁。22     void signalAll();23 }

说明：Condition接口中定义了await、signal函数，用来等待条件、释放条件。之后会详细分析CondtionObject的源码。

3.3 类的属性

1 public abstract class AbstractQueuedSynchronizer 2     extends AbstractOwnableSynchronizer 3     implements java.io.Serializable {     4     // 版本号 5     private static final long serialVersionUID = 7373984972572414691L;     6     // 头结点 7     private transient volatile Node head;     8     // 尾结点 9     private transient volatile Node tail;    10     // 状态11     private volatile int state;    12     // 自旋时间13     static final long spinForTimeoutThreshold = 1000L;14     15     // Unsafe类实例16     private static final Unsafe unsafe = Unsafe.getUnsafe();17     // state内存偏移地址18     private static final long stateOffset;19     // head内存偏移地址20     private static final long headOffset;21     // state内存偏移地址22     private static final long tailOffset;23     // tail内存偏移地址24     private static final long waitStatusOffset;25     // next内存偏移地址26     private static final long nextOffset;27     // 静态初始化块28     static {29         try {30             stateOffset = unsafe.objectFieldOffset31                 (AbstractQueuedSynchronizer.class.getDeclaredField("state"));32             headOffset = unsafe.objectFieldOffset33                 (AbstractQueuedSynchronizer.class.getDeclaredField("head"));34             tailOffset = unsafe.objectFieldOffset35                 (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));36             waitStatusOffset = unsafe.objectFieldOffset37                 (Node.class.getDeclaredField("waitStatus"));38             nextOffset = unsafe.objectFieldOffset39                 (Node.class.getDeclaredField("next"));40 41         } catch (Exception ex) { throw new Error(ex); }42     }43 }

说明：属性中包含了头结点head，尾结点tail，状态state、自旋时间spinForTimeoutThreshold，还有AbstractQueuedSynchronizer抽象的属性在内存中的偏移地址，通过该偏移地址，可以获取和设置该属性的值，同时还包括一个静态初始化块，用于加载内存偏移地址。

3.4 类的构造函数

protected AbstractQueuedSynchronizer() { }

说明：此类构造函数为从抽象构造函数，供子类调用。

3.5 类的核心函数

1. acquire函数

　　该函数以独占模式获取(资源)，忽略中断，即线程在aquire过程中，中断此线程是无效的。源码如下　

1 public final void acquire(int arg) {2     if (!tryAcquire(arg) &&3         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))4          　　selfInterrupt();5 }

由上述源码可以知道，当一个线程调用acquire时，调用方法流程如下。

　说明：

　　① 首先调用tryAcquire函数，调用此方法的线程会试图在独占模式下获取对象状态。此方法应该查询是否允许它在独占模式下获取对象状态，如果允许，则获取它。在AbstractQueuedSynchronizer源码中默认会抛出一个异常，即需要子类去重写此函数完成自己的逻辑。之后会进行分析。

　　② 若tryAcquire失败，则调用addWaiter函数，addWaiter函数完成的功能是将调用此方法的线程封装成为一个结点并放入Sync queue。

　　③ 调用acquireQueued函数，此函数完成的功能是Sync queue中的结点不断尝试获取资源，若成功，则返回true，否则，返回false。

　　由于tryAcquire默认实现是抛出异常，所以此时，不进行分析，之后会结合一个例子进行分析。

　　首先分析addWaiter函数

1 / 添加等待者 2     private Node addWaiter(Node mode) { 3         // 新生成一个结点，默认为独占模式 4         Node node = new Node(Thread.currentThread(), mode); 5         // Try the fast path of enq; backup to full enq on failure 6         // 保存尾结点 7         Node pred = tail; 8         if (pred != null) { // 尾结点不为空，即已经被初始化 9             // 将node结点的prev域连接到尾结点10             node.prev = pred; 11             if (compareAndSetTail(pred, node)) { // 比较pred是否为尾结点，是则将尾结点设置为node 12                 // 设置尾结点的next域为node13                 pred.next = node;14                 return node; // 返回新生成的结点15             }16         }17         enq(node); // 尾结点为空(即还没有被初始化过)，或者是compareAndSetTail操作失败，则入队列18         return node;19 }

说明：addWaiter函数使用快速添加的方式往sync queue尾部添加结点，如果sync queue队列还没有初始化，则会使用enq插入队列中，enq方法源码如下

1 // 入队列 2     private Node enq(final Node node) { 3         for (;;) { // 无限循环，确保结点能够成功入队列 4             // 保存尾结点 5             Node t = tail; 6             if (t == null) { // 尾结点为空，即还没被初始化 7                 if (compareAndSetHead(new Node())) // 头结点为空，并设置头结点为新生成的结点 8                     tail = head; // 头结点与尾结点都指向同一个新生结点 9             } else { // 尾结点不为空，即已经被初始化过10                 // 将node结点的prev域连接到尾结点11                 node.prev = t; 12                 if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点，若是则将尾结点设置为node13                     // 设置尾结点的next域为node14                     t.next = node; 15                     return t; // 返回尾结点16                 }17             }18         }19 }

说明：enq函数会使用无限循环来确保节点的成功插入。

　　现在，分析acquireQueue函数。其源码如下

1 // sync队列中的结点在独占且忽略中断的模式下获取(资源) 2     final boolean acquireQueued(final Node node, int arg) { 3         // 标志 4         boolean failed = true; 5         try { 6             // 中断标志 7             boolean interrupted = false; 8             for (;;) { // 无限循环 9                 // 获取node节点的前驱结点10                 final Node p = node.predecessor(); 11                 if (p == head && tryAcquire(arg)) { // 前驱为头结点并且成功获得锁12                     setHead(node); // 设置头结点13                     p.next = null; // help GC14                     failed = false; // 设置标志15                     return interrupted; 16                 }17                 if (shouldParkAfterFailedAcquire(p, node) &&18                     parkAndCheckInterrupt())19                     interrupted = true;20             }21         } finally {22             if (failed)23                 cancelAcquire(node);24         }25 }

　　说明：首先获取当前节点的前驱节点，如果前驱节点是头结点并且能够获取(资源)，代表该当前节点能够占有锁，设置头结点为当前节点，返回。否则，调用shouldParkAfterFailedAcquire和parkAndCheckInterrupt函数，首先，我们看shouldParkAfterFailedAcquire函数，代码如下

1 // 当获取(资源)失败后，检查并且更新结点状态 2     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 3         // 获取前驱结点的状态 4         int ws = pred.waitStatus; 5         if (ws == Node.SIGNAL) // 状态为SIGNAL，为-1 6             /* 7              * This node has already set status asking a release 8              * to signal it, so it can safely park. 9              */10             // 可以进行park操作11             return true; 12         if (ws > 0) { // 表示状态为CANCELLED，为113             /*14              * Predecessor was cancelled. Skip over predecessors and15              * indicate retry.16              */17             do {18                 node.prev = pred = pred.prev;19             } while (pred.waitStatus > 0); // 找到pred结点前面最近的一个状态不为CANCELLED的结点20             // 赋值pred结点的next域21             pred.next = node; 22         } else { // 为PROPAGATE -3 或者是0 表示无状态,(为CONDITION -2时，表示此节点在condition queue中) 23             /*24              * waitStatus must be 0 or PROPAGATE.  Indicate that we25              * need a signal, but don't park yet.  Caller will need to26              * retry to make sure it cannot acquire before parking.27              */28             // 比较并设置前驱结点的状态为SIGNAL29             compareAndSetWaitStatus(pred, ws, Node.SIGNAL); 30         }31         // 不能进行park操作32         return false;33 }

说明：只有当该节点的前驱结点的状态为SIGNAL时，才可以对该结点所封装的线程进行park操作。否则，将不能进行park操作。再看parkAndCheckInterrupt函数，源码如下

1 // 进行park操作并且返回该线程是否被中断2     private final boolean parkAndCheckInterrupt() {3         // 在许可可用之前禁用当前线程，并且设置了blocker4         LockSupport.park(this);5         return Thread.interrupted(); // 当前线程是否已被中断，并清除中断标记位6 }

说明：parkAndCheckInterrupt函数里的逻辑是首先执行park操作，即禁用当前线程，然后返回该线程是否已经被中断。再看final块中的cancelAcquire函数，其源码如下

1 // 取消继续获取(资源) 2     private void cancelAcquire(Node node) { 3         // Ignore if node doesn't exist 4         // node为空，返回 5         if (node == null) 6             return; 7         // 设置node结点的thread为空 8         node.thread = null; 9 10         // Skip cancelled predecessors11         // 保存node的前驱结点12         Node pred = node.prev;13         while (pred.waitStatus > 0) // 找到node前驱结点中第一个状态小于0的结点，即不为CANCELLED状态的结点14             node.prev = pred = pred.prev;15 16         // predNext is the apparent node to unsplice. CASes below will17         // fail if not, in which case, we lost race vs another cancel18         // or signal, so no further action is necessary.19         // 获取pred结点的下一个结点20         Node predNext = pred.next;21 22         // Can use unconditional write instead of CAS here.23         // After this atomic step, other Nodes can skip past us.24         // Before, we are free of interference from other threads.25         // 设置node结点的状态为CANCELLED26         node.waitStatus = Node.CANCELLED;27 28         // If we are the tail, remove ourselves.29         if (node == tail && compareAndSetTail(node, pred)) { // node结点为尾结点，则设置尾结点为pred结点30             // 比较并设置pred结点的next节点为null31             compareAndSetNext(pred, predNext, null); 32         } else { // node结点不为尾结点，或者比较设置不成功33             // If successor needs signal, try to set pred's next-link34             // so it will get one. Otherwise wake it up to propagate.35             int ws;36             if (pred != head &&37                 ((ws = pred.waitStatus) == Node.SIGNAL ||38                  (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&39                 pred.thread != null) { // （pred结点不为头结点，并且pred结点的状态为SIGNAL）或者 40                                     // pred结点状态小于等于0，并且比较并设置等待状态为SIGNAL成功，并且pred结点所封装的线程不为空41                 // 保存结点的后继42                 Node next = node.next;43                 if (next != null && next.waitStatus <= 0) // 后继不为空并且后继的状态小于等于044                     compareAndSetNext(pred, predNext, next); // 比较并设置pred.next = next;45             } else {46                 unparkSuccessor(node); // 释放node的前一个结点47             }48 49             node.next = node; // help GC50         }51 }

说明：该函数完成的功能就是取消当前线程对资源的获取，即设置该结点的状态为CANCELLED，接着我们再看unparkSuccessor函数，源码如下

1 // 释放后继结点 2     private void unparkSuccessor(Node node) { 3         /* 4          * If status is negative (i.e., possibly needing signal) try 5          * to clear in anticipation of signalling.  It is OK if this 6          * fails or if status is changed by waiting thread. 7          */ 8         // 获取node结点的等待状态 9         int ws = node.waitStatus;10         if (ws < 0) // 状态值小于0，为SIGNAL -1 或 CONDITION -2 或 PROPAGATE -311             // 比较并且设置结点等待状态，设置为012             compareAndSetWaitStatus(node, ws, 0);13 14         /*15          * Thread to unpark is held in successor, which is normally16          * just the next node.  But if cancelled or apparently null,17          * traverse backwards from tail to find the actual18          * non-cancelled successor.19          */20         // 获取node节点的下一个结点21         Node s = node.next;22         if (s == null || s.waitStatus > 0) { // 下一个结点为空或者下一个节点的等待状态大于0，即为CANCELLED23             // s赋值为空24             s = null; 25             // 从尾结点开始从后往前开始遍历26             for (Node t = tail; t != null && t != node; t = t.prev)27                 if (t.waitStatus <= 0) // 找到等待状态小于等于0的结点，找到最前的状态小于等于0的结点28                     // 保存结点29                     s = t;30         }31         if (s != null) // 该结点不为为空，释放许可32             LockSupport.unpark(s.thread);33 }

说明：该函数的作用就是为了释放node节点的后继结点。

　　对于cancelAcquire与unparkSuccessor函数，如下示意图可以清晰的表示。

说明：其中node为参数，在执行完cancelAcquire函数后的效果就是unpark了s结点所包含的t4线程。

　　现在，再来看acquireQueued函数的整个的逻辑。逻辑如下

　　① 判断结点的前驱是否为head并且是否成功获取(资源)。

　　② 若步骤①均满足，则设置结点为head，之后会判断是否finally模块，然后返回。

　　③ 若步骤①不满足，则判断是否需要park当前线程，是否需要park当前线程的逻辑是判断结点的前驱结点的状态是否为SIGNAL，若是，则park当前结点，否则，不进行park操作。

　　④ 若park了当前线程，之后某个线程对本线程unpark后，并且本线程也获得机会运行。那么，将会继续进行步骤①的判断。

2. release

以独占模式释放对象，其源码如下

1 public final boolean release(int arg) { 2         if (tryRelease(arg)) { // 释放成功 3             // 保存头结点 4             Node h = head;  5             if (h != null && h.waitStatus != 0) // 头结点不为空并且头结点状态不为0 6                 unparkSuccessor(h); //释放头结点的后继结点 7             return true; 8         } 9         return false;10 }

说明：其中，tryRelease的默认实现是抛出异常，需要具体的子类实现，如果tryRelease成功，那么如果头结点不为空并且头结点的状态不为0，则释放头结点的后继结点，unparkSuccessor函数已经分析过，不再累赘。

　　对于其他函数我们也可以分析，与前面分析的函数大同小异，所以，不再累赘。

四、示例分析

1. 示例一

借助下面示例来分析AbstractQueuedSyncrhonizer内部的工作机制。示例源码如下

1 package com.hust.grid.leesf.abstractqueuedsynchronizer; 2  3 import java.util.concurrent.locks.Lock; 4 import java.util.concurrent.locks.ReentrantLock; 5  6 class MyThread extends Thread { 7     private Lock lock; 8     public MyThread(String name, Lock lock) { 9         super(name);10         this.lock = lock;11     }12     13     public void run () {14         lock.lock();15         try {16             System.out.println(Thread.currentThread() + " running");17         } finally {18             lock.unlock();19         }20     }21 }22 public class AbstractQueuedSynchonizerDemo {23     public static void main(String[] args) {24         Lock lock = new ReentrantLock();25         26         MyThread t1 = new MyThread("t1", lock);27         MyThread t2 = new MyThread("t2", lock);28         t1.start();29         t2.start();    30     }31 }

运行结果（可能的一种）：

1 Thread[t1,5,main] running2 Thread[t2,5,main] running

结果分析：从示例可知，线程t1与t2共用了一把锁，即同一个lock。可能会存在如下一种时序。

　说明：首先线程t1先执行lock.lock操作，然后t2执行lock.lock操作，然后t1执行lock.unlock操作，最后t2执行lock.unlock操作。基于这样的时序，分析AbstractQueuedSynchronizer内部的工作机制。

　　① t1线程调用lock.lock函数，其函数调用顺序如下，只给出了主要的函数调用。

说明：其中，前面的部分表示哪个类，后面是具体的类中的哪个方法，AQS表示AbstractQueuedSynchronizer类，AOS表示AbstractOwnableSynchronizer类。

　　② t2线程调用lock.lock函数，其函数调用顺序如下，只给出了主要的函数调用。

说明：经过一系列的函数调用，最后达到的状态是禁用t2线程，因为调用了LockSupport.lock。

　　③ t1线程调用lock.unlock，其函数调用顺序如下，只给出了主要的函数调用。

说明：t1线程中调用lock.unlock后，经过一系列的调用，最终的状态是释放了许可，因为调用了LockSupport.unpark。这时，t2线程就可以继续运行了。此时，会继续恢复t2线程运行环境，继续执行LockSupport.park后面的语句，即进一步调用如下。

说明：在上一步调用了LockSupport.unpark后，t2线程恢复运行，则运行parkAndCheckInterrupt，之后，继续运行acquireQueued函数，最后达到的状态是头结点head与尾结点tail均指向了t2线程所在的结点，并且之前的头结点已经从sync队列中断开了。

　　④ t2线程调用lock.unlock，其函数调用顺序如下，只给出了主要的函数调用。

说明：t2线程执行lock.unlock后，最终达到的状态还是与之前的状态一样。

2. 示例二

下面我们结合Condition实现生产者与消费者，来进一步分析AbstractQueuedSynchronizer的内部工作机制。

　　Depot（仓库）类

1 package com.hust.grid.leesf.reentrantLock; 2  3 import java.util.concurrent.locks.Condition; 4 import java.util.concurrent.locks.Lock; 5 import java.util.concurrent.locks.ReentrantLock; 6  7 public class Depot { 8     private int size; 9     private int capacity;10     private Lock lock;11     private Condition fullCondition;12     private Condition emptyCondition;13     14     public Depot(int capacity) {15         this.capacity = capacity;    16         lock = new ReentrantLock();17         fullCondition = lock.newCondition();18         emptyCondition = lock.newCondition();19     }20     21     public void produce(int no) {22         lock.lock();23         int left = no;24         try {25             while (left > 0) {26                 while (size >= capacity)  {27                     System.out.println(Thread.currentThread() + " before await");28                     fullCondition.await();29                     System.out.println(Thread.currentThread() + " after await");30                 }31                 int inc = (left + size) > capacity ? (capacity - size) : left;32                 left -= inc;33                 size += inc;34                 System.out.println("produce = " + inc + ", size = " + size);35                 emptyCondition.signal();36             }37         } catch (InterruptedException e) {38             e.printStackTrace();39         } finally {40             lock.unlock();41         }42     }43     44     public void consume(int no) {45         lock.lock();46         int left = no;47         try {            48             while (left > 0) {49                 while (size <= 0) {50                     System.out.println(Thread.currentThread() + " before await");51                     emptyCondition.await();52                     System.out.println(Thread.currentThread() + " after await");53                 }54                 int dec = (size - left) > 0 ? left : size;55                 left -= dec;56                 size -= dec;57                 System.out.println("consume = " + dec + ", size = " + size);58                 fullCondition.signal();59             }60         } catch (InterruptedException e) {61             e.printStackTrace();62         } finally {63             lock.unlock();64         }65     }66 }

测试类

1 package com.hust.grid.leesf.reentrantLock; 2  3 class Consumer { 4     private Depot depot; 5     public Consumer(Depot depot) { 6         this.depot = depot; 7     } 8      9     public void consume(int no) {10         new Thread(new Runnable() {11             @Override12             public void run() {13                 depot.consume(no);14             }15         }, no + " consume thread").start();16     }17 }18 19 class Producer {20     private Depot depot;21     public Producer(Depot depot) {22         this.depot = depot;23     }24     25     public void produce(int no) {26         new Thread(new Runnable() {27             28             @Override29             public void run() {30                 depot.produce(no);31             }32         }, no + " produce thread").start();33     }34 }35 36 public class ReentrantLockDemo {37     public static void main(String[] args) throws InterruptedException {38         Depot depot = new Depot(500);39         new Producer(depot).produce(500);40         new Producer(depot).produce(200);41         new Consumer(depot).consume(500);42         new Consumer(depot).consume(200);43     }44 }

运行结果（可能的一种）：

produce = 500, size = 500Thread[200 produce thread,5,main] before awaitconsume = 500, size = 0Thread[200 consume thread,5,main] before awaitThread[200 produce thread,5,main] after awaitproduce = 200, size = 200Thread[200 consume thread,5,main] after awaitconsume = 200, size = 0

说明：根据结果，我们猜测一种可能的时序如下

　说明：p1代表produce 500的那个线程，p2代表produce 200的那个线程，c1代表consume 500的那个线程，c2代表consume 200的那个线程。

　　1. p1线程调用lock.lock，获得锁，继续运行，函数调用顺序在前面已经给出。

　　2. p2线程调用lock.lock，由前面的分析可得到如下的最终状态。

　说明：p2线程调用lock.lock后，会禁止p2线程的继续运行，因为执行了LockSupport.park操作。

　　3. c1线程调用lock.lock，由前面的分析得到如下的最终状态。

说明：最终c1线程会在sync queue队列的尾部，并且其结点的前驱结点（包含p2的结点）的waitStatus变为了SIGNAL。

　　4. c2线程调用lock.lock，由前面的分析得到如下的最终状态。

说明：最终c1线程会在sync queue队列的尾部，并且其结点的前驱结点（包含c1的结点）的waitStatus变为了SIGNAL。

　　5. p1线程执行emptyCondition.signal，其函数调用顺序如下，只给出了主要的函数调用。

说明：AQS.CO表示AbstractQueuedSynchronizer.ConditionObject类。此时调用signal方法不会产生任何其他效果。

　　6. p1线程执行lock.unlock，根据前面的分析可知，最终的状态如下。

　　说明：此时，p2线程所在的结点为头结点，并且其他两个线程（c1、c2）依旧被禁止，所以，此时p2线程继续运行，执行用户逻辑。

　　7. p2线程执行fullCondition.await，其函数调用顺序如下，只给出了主要的函数调用。

　　说明：最终到达的状态是新生成了一个结点，包含了p2线程，此结点在condition queue中；并且sync queue中p2线程被禁止了，因为在执行了LockSupport.park操作。从函数一些调用可知，在await操作中线程会释放锁资源，供其他线程获取。同时，head结点后继结点的包含的线程的许可被释放了，故其可以继续运行。由于此时，只有c1线程可以运行，故运行c1。

　　8. 继续运行c1线程，c1线程由于之前被park了，所以此时恢复，继续之前的步骤，即还是执行前面提到的acquireQueued函数，之后，c1判断自己的前驱结点为head，并且可以获取锁资源，最终到达的状态如下。