并发编程系列之深入理解JUC中AQS（AbstractQueuedSynchronizer）

蔚落 2022-04-17 02:09 378阅读 0赞

### 深入理解JUC中AQS ###

*  AbstractQueuedSynchronizer同步器
 *   *  队列
     *  同步状态
     *  Node节点

AQS通过内部实现的FIFO等待队列来实现资源获取线程的等待工作，如果当前线程获取资源失败，AQS则会将当前线程以及等待状态等信息构成一个Node节点，并将其加入等待队列中，同事会阻塞当前线程；当其它获取到资源的线程释放持有的资源时，则会把等待队列中的线程唤醒，使其再次尝试获取对应的资源。

# AbstractQueuedSynchronizer同步器 #

## 队列 ##

首先从源码看一下队列的结构

/** * Head of the wait queue, lazily initialized. Except for * initialization, it is modified only via method setHead. Note: * If head exists, its waitStatus is guaranteed not to be * CANCELLED. */
        // 等待队列头结点
        private transient volatile Node head;
    
        /** * Tail of the wait queue, lazily initialized. Modified only via * method enq to add new wait node. */
        // 等待队列的尾结点
        private transient volatile Node tail;
    
        /** * The synchronization state. */
        // 同步状态
        private volatile int state;

同步队列的结构如下图：  
![同步队列的结构][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXppdGVuZzAyMjg_size_16_color_FFFFFF_t_70]  
  AQS同步器类中的其他方法主要是用于插入节点、释放节点：  
![使用CAS添加尾节点][CAS]  
![使用CAS释放头节点][CAS 1]

## 同步状态 ##

AQS类使用单个int（32位）来保存同步状态，并暴露出getState、setState以及compareAndSet操作来读取和更新这个同步状态。state被声明为volatile，并且通过CAS操作来实现compareAndSetState，使得当且仅当同步状态拥有一个一致的期望值时，才会被院子的设置成新值，这样就达到了同步状态的原子性管理，确保了同步状态的原子性、可见行和有序性。  
  基于AQS实现的类必须根据暴露出的状态相关方法定义tryAcquire和tryRelease方法，以控制acquire和release操作。当同步状态满足时，tryAcquire方法必须返回true，而当新的同步状态允许后续acquire时，tryRelease方法也必须返回true。

/** * Returns the current value of synchronization state. * This operation has memory semantics of a {@code volatile} read. * @return current state value */
        // 获取变量的读取状态
        protected final int getState() { 
            return state;
        }
    
        /** * Sets the value of synchronization state. * This operation has memory semantics of a {@code volatile} write. * @param newState the new state value */
        // 写入变量的同步状态
        protected final void setState(int newState) { 
            state = newState;
        }
    
        /** * Atomically sets synchronization state to the given updated * value if the current state value equals the expected value. * This operation has memory semantics of a {@code volatile} read * and write. * * @param expect the expected value * @param update the new value * @return {@code true} if successful. False return indicates that the actual * value was not equal to the expected value. */
        // 使用CAS设置同步状态，确保线程安全
        protected final boolean compareAndSetState(int expect, int update) { 
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }

## Node节点 ##

static final class Node { 
            /** Marker to indicate a node is waiting in shared mode */
            // 声明共享模式下的等待节点
            static final Node SHARED = new Node();
            /** Marker to indicate a node is waiting in exclusive mode */
            // 声明独占模式下的等待节点
            static final Node EXCLUSIVE = null;
    
            /** waitStatus value to indicate thread has cancelled */
            // waitStatus的常量值--表示线程已取消
            static final int CANCELLED =  1;
            /** waitStatus value to indicate successor's thread needs unparking */
            // waitStatus的常量值--表示后继线程需要取消挂起
            static final int SIGNAL    = -1;
            /** waitStatus value to indicate thread is waiting on condition */
            // waitStatus的常量值--表示线程正在等待条件
            static final int CONDITION = -2;
            /** * waitStatus value to indicate the next acquireShared should * unconditionally propagate */
             // waitStatus的常量值--表示下一个acquireShared应无条件传播
            static final int PROPAGATE = -3;
    
            /** * Status field, taking on only the values: * SIGNAL: The successor of this node is (or will soon be) * blocked (via park), so the current node must * unpark its successor when it releases or * cancels. To avoid races, acquire methods must * first indicate they need a signal, * then retry the atomic acquire, and then, * on failure, block. * CANCELLED: This node is cancelled due to timeout or interrupt. * Nodes never leave this state. In particular, * a thread with cancelled node never again blocks. * CONDITION: This node is currently on a condition queue. * It will not be used as a sync queue node * until transferred, at which time the status * will be set to 0. (Use of this value here has * nothing to do with the other uses of the * field, but simplifies mechanics.) * PROPAGATE: A releaseShared should be propagated to other * nodes. This is set (for head node only) in * doReleaseShared to ensure propagation * continues, even if other operations have * since intervened. * 0: None of the above * * The values are arranged numerically to simplify use. * Non-negative values mean that a node doesn't need to * signal. So, most code doesn't need to check for particular * values, just for sign. * * The field is initialized to 0 for normal sync nodes, and * CONDITION for condition nodes. It is modified using CAS * (or when possible, unconditional volatile writes). */
            // 其值只能为CANCELLED、SIGNAL、CONDITION、PROPAGATE或0
            // 初始值为0
            volatile int waitStatus;
    
            /** * Link to predecessor node that current node/thread relies on * for checking waitStatus. Assigned during enqueuing, and nulled * out (for sake of GC) only upon dequeuing. Also, upon * cancellation of a predecessor, we short-circuit while * finding a non-cancelled one, which will always exist * because the head node is never cancelled: A node becomes * head only as a result of successful acquire. A * cancelled thread never succeeds in acquiring, and a thread only * cancels itself, not any other node. */
            // 前驱节点
            volatile Node prev;
    
            /** * Link to the successor node that the current node/thread * unparks upon release. Assigned during enqueuing, adjusted * when bypassing cancelled predecessors, and nulled out (for * sake of GC) when dequeued. The enq operation does not * assign next field of a predecessor until after attachment, * so seeing a null next field does not necessarily mean that * node is at end of queue. However, if a next field appears * to be null, we can scan prev's from the tail to * double-check. The next field of cancelled nodes is set to * point to the node itself instead of null, to make life * easier for isOnSyncQueue. */
            // 后继节点
            volatile Node next;
    
            /** * The thread that enqueued this node. Initialized on * construction and nulled out after use. */
            // 当前节点的线程，在节点初始化时赋值，使用后为null
            volatile Thread thread;
    
            /** * Link to next node waiting on condition, or the special * value SHARED. Because condition queues are accessed only * when holding in exclusive mode, we just need a simple * linked queue to hold nodes while they are waiting on * conditions. They are then transferred to the queue to * re-acquire. And because conditions can only be exclusive, * we save a field by using special value to indicate shared * mode. */
            // 下一个等待节点
            Node nextWaiter;
    
            /** * Returns true if node is waiting in shared mode. */
            final boolean isShared() { 
                return nextWaiter == SHARED;
            }
    
            /** * Returns previous node, or throws NullPointerException if null. * Use when predecessor cannot be null. The null check could * be elided, but is present to help the VM. * * @return the predecessor of this node */
            final Node predecessor() throws NullPointerException { 
                Node p = prev;
                if (p == null)
                    throw new NullPointerException();
                else
                    return p;
            }
    
            Node() {     // Used to establish initial head or SHARED marker
            }
    
            Node(Thread thread, Node mode) {      // Used by addWaiter
                this.nextWaiter = mode;
                this.thread = thread;
            }
    
            Node(Thread thread, int waitStatus) {  // Used by Condition
                this.waitStatus = waitStatus;
                this.thread = thread;
            }
        }

Node节点的结构如下图：  
![node节点结构][node]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdXppdGVuZzAyMjg_size_16_color_FFFFFF_t_70]: /images/20220417/f0f9dec095814c1890bfa7b8760a7a94.png
[CAS]: /images/20220417/482185a915254fc1bb46b6395078b017.png
[CAS 1]: /images/20220417/87baebdaa09b41a2998734acbdbb8943.png
[node]: /images/20220417/91aa1079caaa4f8c97cabbc064ded3b3.png