Java并发编程-深入Java同步器AQS原理与应用-线程锁必备知识点

以你之姓@ 2023-07-13 10:43 2阅读 0赞

并发编程中我们常会看到AQS这个词，很多朋友都不知道是什么东东，博主经过翻阅一些资料终于了解了，直接进入主题。

# 简单介绍 #

AQS是AbstractQueuedSynchronizer类的缩写，这个不用多说，大家在Eclipse中输入这个类自然会知道此类是java.util.concurrent.locks包下的一个抽象类。为什么需要重点来分析这个抽象类，因为ReentrantLock、Semaphore、CountDownLatch、ReentrantReadWritLock、ThreadPoolExecutor等都是基于AQS来实现的，AQS是J.U.C（java.util.concurrent）工具包的一个基础类。所有说AQS非常重要，学习并发编程必须了解的一个类。

**查看ReentrantLock源码**可以查看到有一个内部抽象类Sync继承了AbstractQueuedSynchronizer类，其他的相关类都是类似的使用方式，因此可以得知AQS都是通过继承来实现其相应的功能，并且是一个内部类。

public class ReentrantLock implements Lock, java.io.Serializable {
       abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = -5179523762034025860L;
            .......
        }
    
        # 非公平锁
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = 7316153563782823691L;
            .....
        }
        
        #公平锁
        static final class FairSync extends Sync {
            private static final long serialVersionUID = 7316153563782823691L;
            .....
        }
    }

**AQS官方注释给出的一个解释如下（此部分如果觉得太多可忽略不看，直接看总结部分）：**

提供一个框架来实现基于先进先出(FIFO)等待队列的阻塞锁和相关同步器(信号量、事件等)。这个类被设计成大多数类型的同步器的有用基础，这些同步器依赖于单个原子\{@code int\}值来表示状态。子类必须定义改变这个状态的受保护的方法，这些方法定义了这个状态对于被获取或释放的对象意味着什么。考虑到这些，这个类中的其他方法执行所有的排队和阻塞机制。子类可以维护其他状态字段，但是只有使用方法\{@link \#getState\}、\{@link \#setState\}和\{@link \#compareAndSetState\}自动更新的\{@code int\}值才会被同步跟踪。

子类应该定义为非公共的内部帮助类，用于实现其封闭类的同步属性。类\{@code AbstractQueuedSynchronizer\}不实现任何同步接口。相反，它定义了\{@link \#acquireInterruptibly\}等方法，这些方法可以被具体的锁和相关的同步器适当地调用来实现它们的公共方法。

此类支持默认独占模式和共享模式中的一种或两种。当以独占模式获取时，其他线程尝试的获取将无法成功。由多个线程获取的共享模式可能(但不一定)成功。这个类不理解这些区别，除了在机械意义上，当一个共享模式获取成功时，下一个等待的线程(如果存在的话)也必须确定它是否也可以获取。在不同模式下等待的线程共享相同的FIFO队列。通常，实现子类只支持其中一种模式，但是这两种模式都可以发挥作用，例如在\{@link ReadWriteLock\}中。只支持独占模式或只支持共享模式的子类不需要定义支持未使用模式的方法。

这个类定义了一个嵌套的\{@link ConditionObject\}类,可以用作\}\{@link条件由子类实现支持独占模式的方法\{@link \# isHeldExclusively\}报告同步是否只对当前线程持有,\{@link \#释放\}调用方法与当前\{@link \# getState\}值完全释放这个对象,\{@link \#获得\},鉴于这个保存的状态值,最终将此对象恢复到其先前获取的状态。任何\{@code AbstractQueuedSynchronizer\}方法都不会创建这样的条件，所以如果不能满足这个约束，就不要使用它。当然，\{@link ConditionObject\}的行为取决于它的同步器实现的语义。

该类提供内部队列的检查、检测和监视方法，以及条件对象的类似方法。可以根据需要使用\{@code AbstractQueuedSynchronizer\}将它们导出到类中，以实现同步机制。

这个类的序列化只存储底层的原子整数维护状态，因此反序列化的对象有空线程队列。需要序列化的典型子类将定义一个\{@code readObject\}方法，该方法在反序列化时将该对象恢复到已知的初始状态。

# **AQS总结** #

**实现原则：**AQS子类应该定义为非公共的内部帮助类，用于实现其封闭类的同步属性，然后通过调用的方式来实现其相应的同步方法；

**需要深入了解的内容：**

先进先出(FIFO)等待队列实现原理是什么？-- AQS基于先进先出(FIFO)等待队列的阻塞锁和相关同步器

state在AQS起什么作用？-- 这个类被设计成大多数类型的同步器的有用基础，这些同步器依赖于单个原子\{@code int\}值来表示状态

AQS怎么实现独占模式锁和共享模式锁，怎么实现锁并释放锁？

# CLH队列实现原理 #

AQS 核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态。如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制 AQS 是用 CLH 队列锁实现的，即将暂时获取不到锁的线程加入到队列中。

> CLH(Craig,Landin,and Hagersten)队列是一个虚拟的双向队列（虚拟的双向队列即不存在队列实例，仅存在结点之间的关联关系）。AQS 是将每条请求共享资源的线程封装成一个 CLH 锁队列的一个结点（Node）来实现锁的分配。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU_size_16_color_FFFFFF_t_70][]

CLH队列结构如下：

![20200311171605399.png][]

红色节点为头结点,可以把它当做正在持有锁的节点，源码中内部类Node就是CLH队列实现的类，在AQS中拥有队列的头(head)和尾(tail)

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;//同步状态

由上可知,它把head和tail设置为了volatile,这两个节点的修改将会被其他线程看到,事实上,我们也主要是通过修改这两个节点来完成入队和出队.

static final class Node {
        //该等待同步的节点处于共享模式
        static final Node SHARED = new Node();
        //该等待同步的节点处于独占模式
        static final Node EXCLUSIVE = null;
    
        //等待状态,这个和state是不一样的:有1,0,-1,-2,-3五个值
        volatile int waitStatus;
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;
    
        volatile Node prev;//前驱节点
        volatile Node next;//后继节点
        volatile Thread thread;//等待锁的线程
        //和节点是否共享有关
        Node nextWaiter;
        //Returns true if node is waiting in shared mode
        final boolean isShared() {
                return nextWaiter == SHARED;
        }

下面解释下waitStatus五个的得含义:

*  CANCELLED(1):该节点的线程可能由于超时或被中断而处于被取消(作废)状态,一旦处于这个状态,节点状态将一直处于CANCELLED(作废),因此应该从队列中移除.
 *  SIGNAL(-1):当前节点为SIGNAL时,后继节点会被挂起,因此在当前节点释放锁或被取消之后必须被唤醒(unparking)其后继结点.
 *  CONDITION(-2) 该节点的线程处于等待条件状态,不会被当作是同步队列上的节点,直到被唤醒(signal),设置其值为0,重新进入阻塞状态.
 *  0:新加入的节点

在锁的获取时，并不一定只有一个线程才能持有这个锁（或者称为同步状态），所以此时有了**独占模式和共享模式**的区别，也就是在Node节点中由nextWait来标识。比如ReentrantLock就是一个独占锁，只能有一个线程获得锁，而WriteAndReadLock的读锁则能由多个线程同时获取，但它的写锁则只能由一个线程持有。这次先介绍独占模式下锁（或者称为同步状态）的获取与释放.这个类使用到了**模板方法设计模式**（具体了解可看后续的设计模式相关博文）：**定义一个操作中算法的骨架，而将一些步骤的实现延迟到子类中。**

# 独占模式获取锁 #

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

该方法首先尝试获取锁( tryAcquire(arg)的具体实现定义在了子类中),如果获取到,则执行完毕,否则通过addWaiter(Node.EXCLUSIVE), arg)方法把当前节点添加到等待队列末尾,并设置为独占模式

private Node addWaiter(Node mode) {
            //把当前线程包装为node,设为独占模式
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            Node pred = tail;
            //如果tail不为空,把node插入末尾
            if (pred != null) {
                node.prev = pred;
                //此时可能有其他线程插入,所以重新判断tail
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);
            return node;
    }
    
    private Node enq(final Node node) {
            for (;;) {
                Node t = tail;
                //此时可能有其他线程插入,所以重新判断tail是否为空
                if (t == null) { // Must initialize
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
    }

如果tail节点为空,执行enq(node);重新尝试,最终把node插入.在把node插入队列末尾后,它并不立即挂起该节点中线程,因为在插入它的过程中,前面的线程可能已经执行完成,所以它会先进行自旋操作acquireQueued(node, arg),尝试让该线程重新获取锁!当条件满足获取到了锁则可以从自旋过程中退出，否则继续。

final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
                    //如果它的前继节点为头结点,尝试获取锁,获取成功则返回           
                    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);
            }
    }

如果没获取到锁,则判断是否应该挂起,而这个判断则得通过它的前驱节点的waitStatus来确定:

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    	int ws = pred.waitStatus;
    	if (ws == Node.SIGNAL)
    		return true;
    	if (ws > 0) {
    		do {
    			node.prev = pred = pred.prev;
    		} while (pred.waitStatus > 0);
    		pred.next = node;
    	} else {       
    		compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    	}
    	return false;
    }

如果前驱节点的waitStatus为:

*  SIGNAL,则返回true表示应该挂起当前线程,挂起该线程,并等待被唤醒,被唤醒后进行中断检测,如果发现当前线程被中断，那么抛出InterruptedException并退出循环.
 *  >0,将前驱节点踢出队列,返回false
 *  <0,也是返回false,不过先将前驱节点waitStatus设置为SIGNAL,使得下次判断时,将当前节点挂起.

最后,我们对获取独占式锁过程对做个总结:

**AQS****的模板方法acquire通过调用子类自定义实现的tryAcquire获取同步状态失败后->将线程构造成Node节点(addWaiter)->将Node节点添加到同步队列对尾(addWaiter)->节点以自旋的方法获取同步状态(acquirQueued)。在节点自旋获取同步状态时，只有其前驱节点是头节点的时候才会尝试获取同步状态，如果该节点的前驱不是头节点或者该节点的前驱节点是头节点单获取同步状态失败，则判断当前线程需要阻塞，如果需要阻塞则需要被唤醒过后才返回。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU_size_16_color_FFFFFF_t_70 1][]

# 独占模式同步状态的释放 #

既然是释放,那肯定是持有锁的该线程执行释放操作,即head节点中的线程释放锁.

AQS中的release释放同步状态和acquire获取同步状态一样，都是模板方法，tryRelease释放的具体操作都有子类去实现，父类AQS只提供一个算法骨架。

public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0){
                unparkSuccessor(h);
            }
            return true;
        }
        return false;
    }
    
    /**如果node的后继节点不为空且不是作废状态,则唤醒这个后继节点,否则从末尾开始寻找合适的节点,如果找到,则唤醒*/
    private void unparkSuccessor(Node node) {
            int ws = node.waitStatus;
            if (ws < 0){
                compareAndSetWaitStatus(node, ws, 0);
            }
            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);
            }
    }

过程:首先调用子类的tryRelease()方法释放锁,然后唤醒后继节点,在唤醒的过程中,需要判断后继节点是否满足情况,如果后继节点不为且不是作废状态,则唤醒这个后继节点,否则从tail节点向前寻找合适的节点,如果找到,则唤醒.

# 共享锁 #

**获取锁的过程：**

1.  当线程调用acquireShared()申请获取锁资源时，如果成功，则进入临界区。
2.  当获取锁失败时，则创建一个共享类型的节点并进入一个FIFO等待队列，然后被挂起等待唤醒。
3.  当队列中的等待线程被唤醒以后就重新尝试获取锁资源，如果成功则**唤醒后面还在等待的共享节点并把该唤醒事件传递下去，即会依次唤醒在该节点后面的所有共享节点**，然后进入临界区，否则继续挂起等待。

**释放锁过程：**

1.  当线程调用releaseShared()进行锁资源释放时，如果释放成功，则唤醒队列中等待的节点，如果有的话。

## 共享锁源码深入分析 ##

基于上面所说的共享锁执行流程，我们接下来看下源码实现逻辑：  
首先来看下获取锁的方法acquireShared()，如下

public final void acquireShared(int arg) {
            //尝试获取共享锁，返回值小于0表示获取失败
            if (tryAcquireShared(arg) < 0)
                //执行获取锁失败以后的方法
                doAcquireShared(arg);
        }

这里tryAcquireShared()方法是留给用户去实现具体的获取锁逻辑的。关于该方法的实现有两点需要特别说明：

**一、该方法必须自己检查当前上下文是否支持获取共享锁，如果支持再进行获取。**

**二、该方法返回值是个重点。其一、由上面的源码片段可以看出返回值小于0表示获取锁失败，需要进入等待队列。其二、如果返回值等于0表示当前线程获取共享锁成功，但它后续的线程是无法继续获取的，也就是不需要把它后面等待的节点唤醒。最后、如果返回值大于0，表示当前线程获取共享锁成功且它后续等待的节点也有可能继续获取共享锁成功，也就是说此时需要把后续节点唤醒让它们去尝试获取共享锁。**

有了上面的约定，我们再来看下doAcquireShared方法的实现：

//参数不多说，就是传给acquireShared()的参数
        private void doAcquireShared(int arg) {
            //添加等待节点的方法跟独占锁一样，唯一区别就是节点类型变为了共享型，不再赘述
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
                    //表示前面的节点已经获取到锁，自己会尝试获取锁
                    if (p == head) {
                        int r = tryAcquireShared(arg);
                        //注意上面说的， 等于0表示不用唤醒后继节点，大于0需要
                        if (r >= 0) {
                            //这里是重点，获取到锁以后的唤醒操作，后面详细说
                            setHeadAndPropagate(node, r);
                            p.next = null;
                            //如果是因为中断醒来则设置中断标记位
                            if (interrupted)
                                selfInterrupt();
                            failed = false;
                            return;
                        }
                    }
                    //挂起逻辑跟独占锁一样，不再赘述
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                //获取失败的取消逻辑跟独占锁一样，不再赘述
                if (failed)
                    cancelAcquire(node);
            }
        }

独占锁模式获取成功以后设置头结点然后返回中断状态，结束流程。而共享锁模式获取成功以后，调用了setHeadAndPropagate方法，从方法名就可以看出除了设置新的头结点以外还有一个传递动作，一起看下代码：

//两个入参，一个是当前成功获取共享锁的节点，一个就是tryAcquireShared方法的返回值，注意上面说的，它可能大于0也可能等于0
        private void setHeadAndPropagate(Node node, int propagate) {
            Node h = head; //记录当前头节点
            //设置新的头节点，即把当前获取到锁的节点设置为头节点
            //注：这里是获取到锁之后的操作，不需要并发控制
            setHead(node);
            //这里意思有两种情况是需要执行唤醒操作
            //1.propagate > 0 表示调用方指明了后继节点需要被唤醒
            //2.头节点后面的节点需要被唤醒（waitStatus<0），不论是老的头结点还是新的头结点
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
                Node s = node.next;
                //如果当前节点的后继节点是共享类型或者没有后继节点，则进行唤醒
                //这里可以理解为除非明确指明不需要唤醒（后继等待节点是独占类型），否则都要唤醒
                if (s == null || s.isShared())
                    //后面详细说
                    doReleaseShared();
            }
        }
    
        private void setHead(Node node) {
            head = node;
            node.thread = null;
            node.prev = null;
        }

最终的唤醒操作也很复杂，专门拿出来分析一下：  
**注：这个唤醒操作在releaseShare()方法里也会调用。**

private void doReleaseShared() {
            for (;;) {
                //唤醒操作由头结点开始，注意这里的头节点已经是上面新设置的头结点了
                //其实就是唤醒上面新获取到共享锁的节点的后继节点
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    //表示后继节点需要被唤醒
                    if (ws == Node.SIGNAL) {
                        //这里需要控制并发，因为入口有setHeadAndPropagate跟release两个，避免两次unpark
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;      
                        //执行唤醒操作      
                        unparkSuccessor(h);
                    }
                    //如果后继节点暂时不需要唤醒，则把当前节点状态设置为PROPAGATE确保以后可以传递下去
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                
                }
                //如果头结点没有发生变化，表示设置完成，退出循环
                //如果头结点发生变化，比如说其他线程获取到了锁，为了使自己的唤醒动作可以传递，必须进行重试
                if (h == head)                   
                    break;
            }
        }

接下来看下释放共享锁的过程：

public final boolean releaseShared(int arg) {
            //尝试释放共享锁
            if (tryReleaseShared(arg)) {
                //唤醒过程，详情见上面分析
                doReleaseShared();
                return true;
            }
            return false;
        }

**注：上面的setHeadAndPropagate()方法表示等待队列中的线程成功获取到共享锁，这时候它需要唤醒它后面的共享节点（如果有），但是当通过releaseShared（）方法去释放一个共享锁的时候，接下来等待独占锁跟共享锁的线程都可以被唤醒进行尝试获取。**

**三、总结**

跟独占锁相比，共享锁的主要特征在于当一个在等待队列中的共享节点成功获取到锁以后（它获取到的是共享锁），既然是共享，那它必须要依次唤醒后面所有可以跟它一起共享当前锁资源的节点，毫无疑问，这些节点必须也是在等待共享锁（这是大前提，如果等待的是独占锁，那前面已经有一个共享节点获取锁了，它肯定是获取不到的）。当共享锁被释放的时候，可以用读写锁为例进行思考，当一个读锁被释放，此时不论是读锁还是写锁都是可以竞争资源的。

笔者的微信公众号，每天一篇好文章：

![70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200311171402335.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU=,size_16,color_FFFFFF,t_70
[20200311171605399.png]: https://img-blog.csdnimg.cn/20200311171605399.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200311211121556.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU=,size_16,color_FFFFFF,t_70
[70]: https://img-blog.csdn.net/20180718011918368?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JlbHZpbmU=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70