【搞定Java并发编程】第15篇：队列同步器AQS源码分析之概要分析

朱雀 2022-04-05 12:17 152阅读 0赞

**AQS系列文章：**

**[1、队列同步器AQS源码分析之概要分析][1_AQS]**

**[2、队列同步器AQS源码分析之独占模式][2_AQS]**

**[3、队列同步器AQS源码分析之共享模式][3_AQS]**

**[4、队列同步器AQS源码分析之Condition接口、等待队列][4_AQS_Condition]**

先推荐两篇不错的博文：

**[1、一行一行源码分析清楚AbstractQueuedSynchronizer][1_AbstractQueuedSynchronizer]**

**[2、深入剖析基于并发AQS的(独占锁)重入锁(ReetrantLock)及其Condition实现原理][2_AQS_ReetrantLock_Condition]**

--------------------

**队列同步器 AbstractQueuedSynchronizer (简称：同步器、AQS)，是用来构建锁或者其他同步组件的基础框架，它使用了一个 int 成员变量表示同步状态，通过内置的 FIFO 队列来完成资源获取线程的排队工作。**

同步器从其名字就可以看出来它是一个抽象类。因此，**同步器的主要使用方式是继承，子类通过继承同步器并实现它的抽象方法来管理同步状态。**

同步状态的改变主要通过同步器提供的三个方法：**getState()、setState(int  newState)** 和 **compareAndSetState(int expect, int update)。**

**子类被推荐定义为自定义同步组件的静态内部类，同步器自身没有实现任何同步接口，它仅仅定义了若干同步状态获取和释放的方法来供自定义同步组件使用。**

**同步器既可以支持独占式地获取同步状态，也可以支持共享式地获取同步状态**，这样就可以方便实现不同类型的同步组件（ReentrantReadWriteLock 和 CountDownLatch）。

*  **同步器与锁的关系：**

同步器是实现锁的关键，在锁的实现中聚合同步器，利用同步器实现锁的语义。

可以这样理解二者的关系：锁是面向使用者的，它定义了使用者与锁交互的接口，隐藏了实现细节；而同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待和唤醒等底层操作。锁和同步器很好地隔离了使用者和实现者所需要关注的领域。

--------------------

# 1、队列同步器的接口与示例 #

同步器的设计是基于模板方法模式的，即使用者需要继承同步器并重写指定的方法，随后将同步器组合在自定义同步组件的实现中，并调用同步器提供的模板方法，而这些模板方法将调用使用者重写的方法。

同步器可重写的方法如下表所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70][]**同步器可重写的方法**

同步器提供的模板方法如下表所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 1][]**同步器提供的模板方法**

上表中的同步器模板方法基本上分为3类：**独占式获取和释放同步状态、共享式获取和释放同步状态**和**查询同步队列中的等待线程情况。**

下面先看一个独占锁的案例：

所谓独占式，即在同一时刻只能有一个线程获取到锁，而其他获取锁的线程只能处于同步队列中等待，只有获取锁的线程释放了锁，后继的线程才能够获取锁。

package zju.com.lock;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.AbstractQueuedSynchronizer;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    
    /**
     * 实现一个独占锁
     */
    public class Mutex implements Lock {
    
    	// 静态内部类，自定义同步器
    	private static class Sync extends AbstractQueuedSynchronizer{
    		
    		// 是否处于独占状态
    		protected boolean isHeldExclusively(){
    			return getState() == 1;
    		}
    		
    		// 当状态为0时获取锁
    		public boolean tryAcquire(int acquires){
    			if(compareAndSetState(0, 1)){
    				setExclusiveOwnerThread(Thread.currentThread());
    				return true;
    			}
    			return false;
    		}
    		
    		// 释放锁，将状态设置为0
    		protected boolean tryRelease(int releases){
    			if(getState() == 0) throw new IllegalMonitorStateException();
    			setExclusiveOwnerThread(null);
    			setState(0);
    			return true;
    		}
    		
    		// 返回一个Condition，每个condition都包含了一个condition队列
    		Condition newCondition(){
    			return new ConditionObject();
    		}
    	}
    	
    	// 仅需要将操作代理到Sync上即可
    	private final Sync sync = new Sync();
    	
    	@Override
    	public void lock() {
    		sync.acquire(1);
    	}
    
    	@Override
    	public void lockInterruptibly() throws InterruptedException {
    		sync.acquireInterruptibly(1);
    	}
    
    	@Override
    	public boolean tryLock() {
    		return sync.tryAcquire(1);
    	}
    
    	@Override
    	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    		return sync.tryAcquireNanos(1, unit.toNanos(time));
    	}
    
    	@Override
    	public void unlock() {
    		sync.release(1);
    	}
    
    	@Override
    	public Condition newCondition() {
    		return sync.newCondition();
    	}
    }

上诉例子中，独占锁Mutex是一个自定义同步组件，它在同一时刻只允许一个线程占有锁。

Mutex中定义了一个静态内部类Sync，该内部类继承了同步器AQS并实现了独占式获取（tryAcquire）和释放同步状态（tryRelease）。在tryAcquire(int  acquires)方法中，如果经过CAS设置成功（同步状态设置为1）则代表获取到了同步状态，而在tryRelease(int  releases)方法中只是将同步状态重置为0。

用户使用Mutex时并不会直接和内部同步器的实现打交道，而是调用Mutex提供的方法。在Mutex的实现中，以获取锁的lock()为例，只需要在方法实现中调用同步器的模板方法acquire(int  args)即可，当前线程调用该方法获取同步状态失败后会被加入到同步队列中等待，这样就大大降低了实现一个可靠自定义同步组件的门槛。

--------------------

# 2、对列同步器AQS的源码分析 #

AbstractQueuedSynchronizer源码有一千多行，但是重复的也比较多，所以不要刚开始的时候被吓到，只要耐着性子去看慢慢的自然能够渐渐领悟。阅读AbstractQueuedSynchronizer源码有几个比较关键的地方需要弄明白，分别是：独占模式和共享模式的区别，节点的等待状态，以及对条件队列的理解。

下面将从AQS的源码分析队列同步器AQS是如何完成线程同步的，主要包括：**同步队列、独占式同步状态获取和释放、共享式同步状态获取与释放**以及**超时获取同步状态**等同步器的核心数据结构与模板方法。

## 同步队列 ##

同步器依赖内部的同步队列（一个FIFO双向队列）来完成同步状态管理，当前线程获取同步状态管理，当前线程获取同步状态失败时，同步器会将当前线程以及等待状态等信息构造为一个节点（Node）并将其加入同步队列，同时会阻塞当前线程，当同步状态释放时，会把首节点中的线程唤醒，使其再次尝试获取同步状态。

同步队列中的节点（Node）用来保存获取同步状态失败的线程引用、等待状态以及前驱后后继节点。节点的属性类型与名称以及描述如下表所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 2][]

节点是构成同步队列的基础，同步器拥有首节点（head）和尾节点（tail），没有成功获取同步状态的线程将会成为节点加入该队列的尾部。同步队列的基本结构如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 3][]

当一个线程成功地获取了同步状态时，其他线程将无法获取到同步状态，转而被构造成为节点并加入到同步队列中，而这个**加入队列的过程必须要保证线程安全，**因此同步器提供了一个基于CAS的设置尾节点的方法：compareAndSetTail(Node  expect,  Node  update)，它需要传递当前线程”认为“的尾节点和当前节点，只有设置成功后，当前节点才正式与之前的尾结点建立连续。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 4][]**节点加入到同步队列**

**同步队列遵循FIFO（先进先出），首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，将会唤醒后继节点，而后继节点将会在获取同步状态成功时将自己设置为首节点。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 5][]**首节点的设置**

设置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能够成功获取到同步状态，因此设置首节点的方法并不需要使用CAS来保证。

*  **Node的源码**

节点类Node是AQS中的一个静态内部类，源码如下：

static final class Node {
            static final Node SHARED = new Node();  // 表示当前线程以共享模式持有锁
            
            static final Node EXCLUSIVE = null;     // 表示当前线程以独占模式持有锁
    
            static final int CANCELLED =  1;        // 表示当前节点已经取消获取锁
           
            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;                 // 当前节点的线程引用
    
            Node nextWaiter;                        // 表示条件队列中的后继节点
    
            // 判断当前节点是否是共享模式
            final boolean isShared() {
                return nextWaiter == SHARED;
            }
    
            // 返回当前节点的前驱节点
            final Node predecessor() throws NullPointerException {
                Node p = prev;
                if (p == null)
                    throw new NullPointerException();
                else
                    return p;
            }
    
            // 构造器1：无参构造器
            Node() {    
            }
    
            // 构造器2：默认构造器
            Node(Thread thread, Node mode) {     // Used by addWaiter
                this.nextWaiter = mode;
                this.thread = thread;
            }
    
            // 构造器3：在条件队列中使用
            Node(Thread thread, int waitStatus) { // Used by Condition
                this.waitStatus = waitStatus;
                this.thread = thread;
            }
        }

其中SHARED和EXCLUSIVE常量分别代表共享模式和独占模式，所谓共享模式是一个锁允许多条线程同时操作，如信号量Semaphore采用的就是基于AQS的共享模式实现的，而独占模式则是同一个时间段只能有一个线程对共享资源进行操作，多余的请求线程需要排队等待，如ReentranLock。

变量waitStatus则表示当前被封装成Node结点的等待状态，共有4种取值CANCELLED、SIGNAL、CONDITION、PROPAGATE。

> **CANCELLED：**值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。
> 
> **SIGNAL：**值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。
> 
> **CONDITION：**值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
> 
> **PROPAGATE：**值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。
> 
> **0状态：**值为0，代表初始化状态。

*  **AQS中的成员变量**

AQS的成员变量只有三个，分别是同步队列头结点引用，同步队列尾结点引用以及同步状态。

注意，这三个成员变量都使用了 volatile 关键字进行修饰，这就确保了多个线程对它的修改都是内存可见的。整个类的核心就是这个同步状态，可以看到同步状态其实就是一个int型的变量，大家可以把这个同步状态看成一个密码锁，而且还是从房间里面锁起来的密码锁，state 具体的值就相当于密码控制着密码锁的开合。

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
    
        static final class Node{
            // ...
        }
    
        ...
    
        // 同步队列的头结点
        private transient volatile Node head; 
    
        // 同步队列的尾结点
        private transient volatile Node tail;
    
        // 同步状态
        private volatile int state;
    
        // 获取同步状态
        protected final int getState() {
            return state;
        }
    
        // 设置同步状态
        protected final void setState(int newState) {
            state = newState;
        }
    
        // 以CAS方式设置同步状态
        protected final boolean compareAndSetState(int expect, int update) {
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    
        ...
    }

--------------------

**AQS系列文章：**

**[1、队列同步器AQS源码分析之概要分析][1_AQS]**

**[2、队列同步器AQS源码分析之独占模式][2_AQS]**

**[3、队列同步器AQS源码分析之共享模式][3_AQS]**

**[4、队列同步器AQS源码分析之Condition接口、等待队列][4_AQS_Condition]**

## 参考及推荐： ##

**[1、AbstractQueuedSynchronizer源码分析之概要分析][1_AbstractQueuedSynchronizer 1]**

**2、[AbstractQueuedSynchronizer源码分析之独占模式][AbstractQueuedSynchronizer]**

**3、[AbstractQueuedSynchronizer源码分析之共享模式][AbstractQueuedSynchronizer 1]**

**4、[AbstractQueuedSynchronizer源码分析之条件队列][AbstractQueuedSynchronizer 2]**

[1_AQS]: https://blog.csdn.net/pcwl1206/article/details/84972609
[2_AQS]: https://blog.csdn.net/pcwl1206/article/details/84980812
[3_AQS]: https://blog.csdn.net/pcwl1206/article/details/84983118
[4_AQS_Condition]: https://blog.csdn.net/pcwl1206/article/details/84984713
[1_AbstractQueuedSynchronizer]: https://blog.csdn.net/a724888/article/details/60878991
[2_AQS_ReetrantLock_Condition]: https://blog.csdn.net/javazejian/article/details/75043422
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70]: /images/20220405/9ff7313455454cdea2b31c1201f2134b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 1]: /images/20220405/5ffbb9a9f913432392bb5a8b5dce2226.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 2]: /images/20220405/80bb1348f4b241b9a43b8ccfbd9ec532.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 3]: /images/20220405/83d84cfa2e0141c4a28bd5dcd687f358.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 4]: /images/20220405/d313edec57b74e78a010d52fca6e1ce0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 5]: /images/20220405/944fd7102ec8454eac7a59d28e8ce29f.png
[1_AbstractQueuedSynchronizer 1]: https://mp.weixin.qq.com/s/RzD-_SAo3f68lS_05H4B2g
[AbstractQueuedSynchronizer]: https://mp.weixin.qq.com/s/ORoB7yj5oA35FCtuR8gYUA
[AbstractQueuedSynchronizer 1]: https://mp.weixin.qq.com/s/BGB1_UTujwmlQAFIwx4zXQ
[AbstractQueuedSynchronizer 2]: https://mp.weixin.qq.com/s/95wZVLSGbY9ubu3Ot-eAvQ