轻松理解AQS框架 |不会有人看了不懂吧

青旅半醒 2022-10-31 13:47 227阅读 0赞

> 本文作者：可乐可乐可，博主个人主页：[可乐可乐可的个人主页][Link 1]

# 轻松理解AQS框架 #

> 本文需要以下知识铺垫：Java、临界区、信号量、锁

> AQS（AbstractQueuedSynchronizer，抽象队列同步器）是Java中重入锁ReentrantLock、读写锁、信号量的实现基石。
> 
> 学会、了解AQS框架对了解Java锁有很大的帮助

说的比唱的好听，AQS源码下来2k+行，这是人干的活吗？

为了解决大家AQS不了解以及看了忘，忘了看的恶性循环，下面将带领大家从简到繁，一步步的学会AQS框架。

> 本文中涉及的代码以及做好了中文的注释，带伙可以访问我的github仓库拉下来看
> 
> github仓库地址：[Jirath-Liu][]

# AQS是啥 #

各位Java开发者必然会了解一个类，叫ReentrantLock。

在早期使用ReentrantLock效率是远远超过synchronized关键字的，现在差距一步步缩小了。

不知道有没有人点开过ReentrantLock的源码一探究竟？

ReentrantLock内部真正起作用的是Sync类，ReentrantLock所有跟锁有关的方法都调用了Sync的方法来实际实现。

而Sync的父类正是我们的主角——AbstractQueuedSynchronizer

![image-20210221221321504][]

public void lock() {
        
        sync.lock();
    }
    public void lockInterruptibly() throws InterruptedException {
        
            sync.lockInterruptibly();
    }
    public boolean tryLock() {
        
            return sync.tryLock();
    }
    ....

关于ReentrantLock这里就不在啰嗦了，如果有想了解的可以留言，给大伙安排上

我们现在关心的，是里面的这个Sync——AQS的实现类

Sync，或者说AQS的实现类，究竟做了什么，达到了加锁的目的？

加锁大家应该都知道是什么概念吧，信号想必比各位也应该了解（不清楚的先去搜搜

加锁的实质就是信号量，若有线程占用了某个资源，就在信号量进行标记，其他线程就了解这段临界区是被占用的。

我们AQS的原理其实就是信号量机制，Sync的机制如下图

![image-20210221222707969][]

这其中的过程都是由AQS来实现的，Sync编写了一些核心的判断来定制。

![image-20210221222957211][]

上图为Sync的源码，acquire是AQS的方法。

扯了这么半天，想必对AQS有个模糊的认识了：

一个实现了信号量，等待，抢夺锁的轻量级框架。

> **AbstractQueuedSynchronizer**
> 
> --------------------
> 
> 提供一个框架来实现依赖于**先进先出（FIFO）等待队列的阻塞锁**和**相关的同步器**（信号量，事件等）。
> 
> 此类旨在为大多数依赖单个原子int值表示状态的同步器提供有用的基础。
> 
> 子类必须定义更改此状态的受保护方法，并定义该状态对于获取或释放此对象而言意味着什么。

# 如何使用AQS来构建自己的锁？ #

我们先学会用AQS，再探知AQS的原理，总得先会跑，再想怎么跑步省力气吧

AQS框架的大佬给我们提供了四个需要我们实现的接口：

![image-20210221223712534][]

这几个方法都默认直接抛出异常：UnsupportedOperationException，需要子类继承来重写。

这四个方法都是干啥的？

AQS使用了**模板方法**的设计模式，这四个方法除了编写后直接使用外，更会被框架的其他方法调用。只要我们按照规矩老老实实的编写好这四个方法，就能得到一个出自自己之手的高效的轻量的锁。

这四个方法的功能就在下面了

// Main exported methods
    
    /**
     * 尝试以独占模式进行获取。
     *
     * 此方法应查询对象的状态是否允许以独占模式获取它，如果允许则获取它。
     * 此方法始终由执行获取的线程调用。
     * 如果此方法报告失败，则acquire方法可以将线程排队（如果尚未排队），
     * 直到其他某个线程释放该信号为止。
     *
     * 这可以用来实现方法Lock.tryLock() 。
     * 默认实现抛出UnsupportedOperationException 。
     */
    protected boolean tryAcquire(int arg) {
        
        throw new UnsupportedOperationException();
    }
    
    /**
     * 尝试设置状态以反映排他模式下的发布。
     * 始终由执行释放的线程调用此方法。
     * 默认实现抛出UnsupportedOperationException 。
     */
    protected boolean tryRelease(int arg) {
        
        throw new UnsupportedOperationException();
    }
    
    /**
     * 尝试以共享模式进行获取。
     * 此方法应查询对象的状态是否允许以共享模式获取对象，如果允许则获取对象。
     * 此方法始终由执行获取的线程调用。
     * 如果此方法报告失败，则acquire方法可以将线程排队（如果尚未排队），直到其他某个线程释放该信号为止。
     * 默认实现抛出UnsupportedOperationException 。
     */
    protected int tryAcquireShared(int arg) {
        
        throw new UnsupportedOperationException();
    }
    
    /**
     * 尝试设置状态以反映共享模式下的发布。
     * 始终由执行释放的线程调用此方法。
     * 默认实现抛出UnsupportedOperationException 。
     */
    protected boolean tryReleaseShared(int arg) {
        
        throw new UnsupportedOperationException();
    }
    
    /**
     * 如果仅相对于当前（调用）线程保持同步，则返回true 。
     * 每次调用非等待的AbstractQueuedSynchronizer.ConditionObject方法时，都会调用此方法。
     * （等待方法改为调用release 。）
     * 默认实现抛出UnsupportedOperationException 。
     * 此方法仅在AbstractQueuedSynchronizer.ConditionObject方法内部内部调用，
     * 因此如果不使用条件，则无需定义。
     *
     */
    protected boolean isHeldExclusively() {
        
        throw new UnsupportedOperationException();
    }

当然，只拿了这四个方法，肯定是一脸懵逼的，AQS框架提供了很多的方法供子类使用，这些方法都是模板方法，final类型

![image-20210222093629615][]

![image-20210222092649114][]

大致有这些方法：

1.  获取独占锁，以及各种姿势来获取（超时，响应中断，尝试等等），这些方法命名为acquire
2.  获取共享锁，以及各种姿势来获取（超时，响应中断，尝试等等），这些方法命名为acquireShared
3.  释放锁，包括释放独占锁和释放共享锁，释放锁是不处理线程争夺问题的
4.  对等待（Wait）的操作
5.  对AQS的感知，
    
    1.  是否有排队，目标线程是不是在排队
    2.  设置与获取当前的线程（在父类AbstractOwnableSynchronizer中实现），
    3.  CAS设置信号量（compareAndSetState，本人认为这里称为信号量更合适），获取信号量，

总结下来AQS给用户提供了CAS获取锁，修改信号量，对AQS内部感知，锁操作的方法

这些方法一次堆上来就会眼花缭乱，我们可以从ReentrantLock中获取如何使用这些方法。

# ReentrantLock中如何使用AQS #

ReentrantLock中分为公平和非公平锁，这里的公平意思是在获取锁的时候，非公平的锁会直接尝试进行获取，而公平的锁会先看看自己会不会排在第一个，这意味着一个线程释放锁后再次获取锁，成功的几率会较其他线程高些。

我们先看公平锁的实现，来理解如何使用AQS

![image-20210222095912743][]

abstract static class Sync extends AbstractQueuedSynchronizer {
        
        private static final long serialVersionUID = -5179523762034025860L;
        
        final boolean tryLock() {
        
            Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
        
                if (compareAndSetState(0, 1)) {
        
                    setExclusiveOwnerThread(current);
                    return true;
                }
                //重入操作
            } else if (getExclusiveOwnerThread() == current) {
        
                if (++c < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(c);
                return true;
            }
            return false;
        }
    
        abstract boolean initialTryLock();
    
        final void lock() {
        
            if (!initialTryLock())
                acquire(1);
        }
    
        final void lockInterruptibly() throws InterruptedException {
        
            if (Thread.interrupted())
                throw new InterruptedException();
            if (!initialTryLock())
                acquireInterruptibly(1);
        }
    
        final boolean tryLockNanos(long nanos) throws InterruptedException {
        
            if (Thread.interrupted())
                throw new InterruptedException();
            return initialTryLock() || tryAcquireNanos(1, nanos);
        }
    
        protected final boolean tryRelease(int releases) {
        
            int c = getState() - releases;
            if (getExclusiveOwnerThread() != Thread.currentThread())
                throw new IllegalMonitorStateException();
            boolean free = (c == 0);
            if (free)
                setExclusiveOwnerThread(null);
            setState(c);
            return free;
        }
    
        protected final boolean isHeldExclusively() {
        
    
            return getExclusiveOwnerThread() == Thread.currentThread();
        }
    
        final ConditionObject newCondition() {
        
            return new ConditionObject();
        }
    
        final Thread getOwner() {
        
            return getState() == 0 ? null : getExclusiveOwnerThread();
        }
    
        final int getHoldCount() {
        
            return isHeldExclusively() ? getState() : 0;
        }
    
        final boolean isLocked() {
        
            return getState() != 0;
        }
    
        private void readObject(java.io.ObjectInputStream s)
                throws java.io.IOException, ClassNotFoundException {
        
            s.defaultReadObject();
            setState(0); // reset to unlocked state
        }
    }
    
    static final class FairSync extends Sync {
        
            private static final long serialVersionUID = -3000897897090466540L;
    
            final boolean initialTryLock() {
        
                Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
        
                    if (!hasQueuedThreads() && compareAndSetState(0, 1)) {
        
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                    //重入
                } else if (getExclusiveOwnerThread() == current) {
        
                    if (++c < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(c);
                    return true;
                }
                return false;
            }
    
            protected final boolean tryAcquire(int acquires) {
        
                if (getState() == 0 && !hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
        
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
        }

通过上文，我们得知使用AQS，需要重写三个方法，第四个方法若需要Condition也需要重写，我们在ReentrantLock中找找这四个方法，因为ReentrantLock是独占锁，所以他没有重写acquireShare方法

## ReentrantLock重写的方法如下 ##

从这几个方法中，我们能看出，ReentrantLock为使用AQS所做的工作

abstract static class Sync extends AbstractQueuedSynchronizer {
        
       
        protected final boolean tryRelease(int releases) {
        
            int c = getState() - releases;
                //只有拥有者才能进行释放
                if (getExclusiveOwnerThread() != Thread.currentThread())
                    throw new IllegalMonitorStateException();
                //若free为0，表示锁解开，释放线程标记，释放信号量
                boolean free = (c == 0);
                if (free)
                    setExclusiveOwnerThread(null);
            	//因为此时还拥有锁，所以不用考虑抢夺、并发问题
                setState(c);
                return free;
        }
    
        //直接判断是不是自己就行了
        protected final boolean isHeldExclusively() {
        
            return getExclusiveOwnerThread() == Thread.currentThread();
        }
    
    }
    
    static final class FairSync extends Sync {
        
    		//公平
            protected final boolean tryAcquire(int acquires) {
        
                /**
                 *  尝试进行抢夺锁，只有这几个条件满足时，才算成功，并且顺序很重要
                 *  1. 信号量为0
                 *  2. 前面没有排队的线程（公平锁才有的判断
                 *  3. 尝试使用CAS抢夺并成功
                 */
                if (getState() == 0 && !hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
        
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
        }

接着，我们注意看三个方法tryLock，lock，initialTryLock这些是ReentrantLock使用的入口方法，从这些方法我们可以了解AQS是怎么使用的

final boolean tryLock() {
        
            Thread current = Thread.currentThread();
            int c = getState();
            //若c不为0，表示有人占用，若为自己占有，就走重入逻辑
            if (c == 0) {
        
                //尝试抢夺锁
                if (compareAndSetState(0, 1)) {
        
                    //cas修改成功，就表示获得了锁，标记自己就行了
                    setExclusiveOwnerThread(current);
                    return true;
                }
                //重入操作
            } else if (getExclusiveOwnerThread() == current) {
        
                if (++c < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(c);
                return true;
            }
            return false;
        }
    
        abstract boolean initialTryLock();
    
        final void lock() {
        
            //先尝试轻量加锁，使用CAS尝试抢夺或重入
            if (!initialTryLock())
                //直接获取锁，不成功就排队等待
                //1是信号量增加的值
                acquire(1);
        }
    
    //在FairSync中
    	final boolean initialTryLock() {
        
                Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
        
                    //公平锁需要判断hasQueuedThreads()
                    if (!hasQueuedThreads() && compareAndSetState(0, 1)) {
        
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                    //重入
                } else if (getExclusiveOwnerThread() == current) {
        
                    if (++c < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(c);
                    return true;
                }
                return false;
            }

## 总结一下ReentrantLock对AQS的使用 ##

AQS中，锁的占用是使用信号量来标记的，AQS实现了锁何时去争夺，以及排队等逻辑，

说到底AQS还是一个框架，而核心争夺锁的部分是由我们编写的，

AQS提供了各种争夺锁会使用的工具类，我们**只需要编写争夺一次的代码**，AQS会帮我们进行重试，阻塞的操作。

# AQS原理是什么，AQS做了什么 #

> 下面就是源码模式，为了帮助各位看懂这2k行的代码，我先做了几个图，我们看图来了解详细的过程。

AQS独占锁整体的思维是：

1.  使用CAS修改信号量的方式争夺锁
2.  若没抢到，就进入排队
3.  队列内的节点若发现前面的节点状态为Signal，就会进入阻塞状态
4.  队列的第一个节点会不停的查询锁状态尝试进行争夺
5.  锁的释放将修改自己状态为0（防止后续节点阻塞），唤醒后续节点进行争夺

AQS在使用时的核心是两个方法：acquire、release

这两个方法的核心就是AQS的核心，可能有点难以接受，但还是建议看一看这个流程图，心中有个大概。

![AQS acquire][]

![AQS release][]

这两个图如果你大概了解了，看看下面这个ABC三个线程的例子

> 非常推荐点开源码去看

1.  线程A获取锁，进入执行状态
2.  线程B进来，抢夺失败，进入排队，发现Head为空，创建了一个Head（state为0），因为Head状态为0，自己在第一个抢夺位，再次进行抢夺
3.  线程B尝试获取锁再次失败，将Head标记为Signal，再次尝试
4.  线程B在第二次尝试中依然失败，检查到Head为Signal，进入休眠
5.  线程C进入，发现锁抢夺失败，将自己封装为一个新节点挂在了线程C后，因为C不是第一个位置，所以不抢夺，检测到线程B是0状态，但是自己没有获得锁，于是标记B为Signal，再次尝试
6.  C再次尝试，发现自己不是第一位，此时B为Signal，C进入休眠。
7.  线程A释放锁，线程B被唤醒开始抢夺，成功后将Head更换为自己，并执行自己的流程
8.  线程B释放锁，线程C被唤醒，将Head更换为C，执行自己的流程

如果上述的流程对你来说已经能够摸清了，下面我们再逐步分析AQS中的方法

## acquire()源码实现 ##

首先，尝试获取锁，失败则加入队列

//先尝试获取锁，成功直接返回，失败则封装线程加入队列
    public final void acquire(int arg) {
        
        if (!tryAcquire(arg) &&
            //失败则封装线程加入队列
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

进入一个死循环，使用CAS去争夺锁，每一轮循环都要检查自己是不是应该阻塞

在检查的过程中可能修改前者的状态

/**
     * 以排他的不间断模式获取已在队列中的线程。
     * 用于条件等待方法以及获取。
     *
     * 当一个线程执行完任务后，不会删除自己，而是保持在Head上
     * 此时后续节点用cas进行争夺，成功后更新head
     */
    final boolean acquireQueued(final Node node, int arg) {
        
        boolean failed = true;
        try {
        
            boolean interrupted = false;
            for (;;) {
        
                //获取之前的节点
                final Node p = node.predecessor();
                //如果p==Head表示自己在队列的第一位，
                // tryAcquire()为尝试获取锁，为子类实现，若成功则表示抢到了锁的权限（CAS操作）
                if (p == head && tryAcquire(arg)) {
        
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // 若需要阻塞，则进行阻塞操作，并设置interrupt为true，线程将在这里阻塞，
                // 直到有锁的释放，才会唤醒第一个等待的。
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
        
            //若发生了异常，导致添加了队列，但是异常弹出了
            //这里需要删除创建的节点
            if (failed)
                cancelAcquire(node);
        }
    }

检查自己是不是应该阻塞，Node信号在这里更新

这个方法有意思的地方是

1.  要是前者状态是取消，就会继续向前访问并不断更新前置节点（这会造成GC回收这个取消的节点），最后断掉这个废掉的链（也可能是一个节点）
2.  要是前者是0（默认状态）就会修改前者为signal，然后返回，再次尝试，在下次访问这里的时候再阻塞。

/**
     *
     * 判断当前这个节点是不是应该阻塞起来
     *
     * 检查并更新无法获取的节点的状态。
     * 如果线程应阻塞，则返回true。
     * 这是所有采集循环中的主要信号控制。
     * 要求pred == node.prev。
     *
     * 前者为signal，阻塞
     * 前者为cancel，更新前者
     * 其他情况：更新前者为signal(更新后再一轮尝试，还没有锁
     * 就会再次访问这里并阻塞了)
     *
     */
    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 {
        
            /*
             * waitStatus一定为0或PROPAGATE。
             * 表示我们需要一个信号，但不要阻塞。
             * 呼叫者将需要重试以确保在阻塞前无法获取
             */
            //尝试修改状态位signal,下次访问时就会阻塞
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

## release()源码实现 ##

/**
     * 解锁，独占锁解锁
     * 以独占模式发布。
     * 调用子类实现的tryRelease，然后唤醒下一个等待的节点
     * 如果{@link #tryRelease}返回true，则通过解锁一个或多个线程来实现。
     * 此方法可用于实现方法{@link Lock#unlock}.
     */
    public final boolean release(int arg) {
        
        //尝试解锁，调用的是子类的实现，
        //注意子类编写这里时，独占锁要判断能不能解锁
        if (tryRelease(arg)) {
        
            Node h = head;
            //若有人在等待锁，就唤醒第一个节点
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

唤醒后继节点，

这里有两个细节：

1.  传参是Head，在本方法中若Head不是0状态就会更新到0状态，更新到0状态后，这个方法其他线程就不会访问这个方法了。
2.  查询可用节点是从后向前查找的，因为前面的节点要是null，是没有后置节点记录的。

/**
     * 唤醒节点的后继者（如果存在）。
     *
     * @param node the node
     */
    private void unparkSuccessor(Node node) {
        
        /*
         * 如果状态是否定的（即可能需要信号），请尝试清除以预期发出信号。
         * 如果失败或等待线程更改状态，则可以。
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
    
        /*
         * 释放线程将保留在后续线程中，该线程通常只是下一个节点。
         * 但是，如果已取消或明显为空，请从尾部向后移动以找到实际的未取消后继。
         */
        Node s = node.next;
        //waitStatus表示该线程已经取消（只有cancel是大于0的）
        if (s == null || s.waitStatus > 0) {
        
            //若后继节点是空的，或者这个节点已经被取消了，那么需要尝试寻找一个合适的节点
            s = null;
            //从后往前遍历，不断更新s（要唤醒的线程）
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            //唤醒一个线程（unpark方法是将线程从阻塞状态唤醒）
            LockSupport.unpark(s.thread);
    }

以上，就是AQS中独占锁的原理，关于共享锁，且听下回分解

> 关注博主，不迷路：[可乐可乐可的个人主页][Link 1]  
> 若文章对你有助，求赏一键三连

![在这里插入图片描述][20210222233119418.gif_pic_center]

[Link 1]: https://blog.csdn.net/weixin_44494373
[Jirath-Liu]: https://github.com/Jirath-Liu/JavaSourceCode
[image-20210221221321504]: /images/20221024/3a4d68e8b6b7456f996bf3df704adc33.png
[image-20210221222707969]: /images/20221024/120b639e236240b6ac0d9122bf53ebe2.png
[image-20210221222957211]: /images/20221024/8ccd53d1d7f7451b8e9eee0225f7514a.png
[image-20210221223712534]: /images/20221024/ce6013171eb74f399aceed93d9791161.png
[image-20210222093629615]: /images/20221024/d6cd2e01314341068ee16ceb34042fdf.png
[image-20210222092649114]: /images/20221024/d82c4477970e4337b4c966320d5b5d47.png
[image-20210222095912743]: /images/20221024/ed55fad6927e450fb714207cf0d928b2.png
[AQS acquire]: /images/20221024/954ca8c4eab0490b8819048715275ae4.png
[AQS release]: /images/20221024/8c7ed9c41b904dd2be461a3ca42d9115.png
[20210222233119418.gif_pic_center]: /images/20221024/bcbd4ad79b234959903c222a2c077e2d.png