并发编程 - 浅谈 AQS 源码

阳光穿透心脏的1/2处 2023-01-23 03:55 172阅读 0赞

#  #

**目录**

一、ReentrantLock

二、AQS

1.和ReentrantLock的关系

2.AQS队列同步器源码分析

同步队列：

独占式同步队列状态获取和释放：

释放锁并且唤醒一下一个处于part挂起状态的线程：

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

# 一、ReentrantLock  #

在Lock接口出现之前，Java程序主要是靠Synchronized关键字来实现锁的功能，但是在Java 5之后在并发包中增加了Lock接口来实现锁的功能，其中ReentrantLock就是Lock接口的一个比较常用的实现类

相比Synchronized（Synchronized的学习可以看本系列博客“”）来说，ReentrantLock有跟多的优势：

1.  释放锁方面： synchronized 只有在异常发生或者同步块执行完之后才会释放锁 但是ReentrantLock可以通过 unlock（）来灵活的释放锁资源
2.  灵活性： ReentrantLock更加灵活
3.  可以尝试非阻塞的获取锁，利用tryLock() 如果获取成功就返回true，否则返回false，不会一直堵塞
4.  中断的获取锁
5.  超时获取锁

# 二、AQS #

## 1.和ReentrantLock的关系 ##

在ReentrantLock聚合了同步器，利用同步器来真实的实现了锁的语义，也可以说ReentrantLock是面向使用者的，同步器是锁的真实实现

结构图：

![image.png][]

*  Lock是锁的统一接口，ReentrantLock是Lock的一个具体实现类
 *  Sync是ReentrantLock的静态内部类，该类继承了  AbstractQueuedSynchronizer
 *  NofaitSync和FaitSync是Sync的两个子类

整理来看AQS使用了模板方法,该图是本人对于AQS使用整理的模板方法关系图，关系读者于评论区一起讨论

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70][]

## 2.AQS队列同步器源码分析 ##

### `同步队列：` ###

同步队列是一个FIFO（先进先出）的双向队列，当线程获取锁资源失败的时候会变为NODE节点，并且加入到同步队列中，同时会堵塞当前线程，当有线程释放锁资源的时候，会唤醒同步队列的首节点

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 1][]

**1. 线程获取锁失败，线程加入队列：**

当有新的线程获取锁资源失败的时候会加入到队列的尾，但是由于是多线程执行的，可能会有多个线程同时需要加入到同步队列的尾，所以我们这个时候利用CAS来保证线程安全，他需要传递当前线程认为的尾节点和当前节点，否则会由于多个线程共同插入，队列混乱。

CompareAndSetTail(Node expect,Node update)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 2][]

**2. 线程获取到锁，并将释放锁的节点移除同步队列：**

首节点是获取锁成功的节点,首节点的线程在释放锁时,会唤醒后续节点,而后继节点在成功获取到锁后,会把自己设置成首节点,设置首节点是由获取锁成功的线程来完成的,由于只有一个线程能成功获取到锁,所以设置首节点不需要`CAS`

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 3][]

###  独占式同步队列状态获取和释放：  ###

//lock方法先通过CAS尝试将同步状态(AQS的state属性)从0修改为1。
    //若直接修改成功了，则将占用锁的线程设置为当前线程       
    final void lock() {
                if (compareAndSetState(0, 1))
                    //用来保存当前占用同步状态的线程。
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    //独占式获取同步状态，如果获取成功就直接返回，否则加入同步队列
                    acquire(1);
            }

**解释：** 同步队列中维护了一个 state 变量，当state 变量为0的时候表示当前锁没有被获取，当有一个线程想要获取到锁的时候，首先会通过CAS去判断当前state是否为0 ，如果为0就抢占锁资源，设置当前线程为同步状态的线程，因为ReentrantLock是支持可重入的，所以当某一个线程判断当前状态不为0，他是有机会继续获取同步状态的（同一个线程可以重入）

**看acquire方法：**

/**
    * tryAcquire方法尝试获取锁，如果成功就返回，
    * 如果不成功，则把当前线程和等待状态信息构适成一个Node节点，
    * 并将结点放入同步队列的尾部。然后为同步队列中的当前节点循环等待获取锁，直到成功
    *
    */
     public final void acquire(int arg) {
            //这里调用的是被子类重写的方法,如果没有获取成功
            if (!tryAcquire(arg) &&
                    //将线程转变为NODE节点，并且加入到同步队列中
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    
                selfInterrupt();
        }

**tryAcquire()方法：**

tryAcquire方法是AQS提供的一个可重写的方法，被ReentrantLock的重写成了公平和非公平锁，我们默认走非公平锁.

final boolean nonfairTryAcquire(int acquires) {
                //获取当前线程
                final Thread current = Thread.currentThread();
                //获取状态
                int c = getState();
                //如果状态为0表示没有被加锁
                if (c == 0) {
                    if (compareAndSetState(0, acquires)) {
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                //如果发现当前加锁的是当前线程，那么就可以重入
                else if (current == getExclusiveOwnerThread()) {
                    //将重入次数+1
                    int nextc = c + acquires;
                    if (nextc < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
         }

addWaiter(Node.EXCLUSIVE), arg)方法：

private Node addWaiter(Node mode) {
                Node node = new Node(Thread.currentThread(), mode);
                // Try the fast path of enq; backup to full enq on failure
                Node pred = tail;
                //如果当前同步队列不为空
                if (pred != null) {
                  
                    node.prev = pred;
                    //从尾部加入节点
                    if (compareAndSetTail(pred, node)) {
                        pred.next = node;
                        return node;
                    }
                }
                //构建同步队列空的首节点
                enq(node);
                return node;
            }

首先创建一个Node对象，Node中包含了当前线程和Node模式(这时是排他模式)。tail是AQS的中表示同步队列队尾的属性，刚开始为null，所以进行enq(node)方法，从字面可以看出这是一个入队操作，来看下具体入队细节

private Node enq(final Node node) {
            for (;;) {
                Node t = 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;
                    }
                }
            }
        }

**解析：**

方法体是一个死循环，本身没有锁，可以多个线程并发访问，假如某个线程进入方法，此时head, tail都为null, 进入if(t==null)区域，从方法名可以看出这里是用CAS的方式创建一个空的Node作为头结点，因为此时队列中只一个头结点，所以tail也指向它，第一次循环执行结束。注意这里使用CAS是防止多个线程并发执行到这儿时，只有一个线程能够执行成功，防止创建多个同步队列。

进行第二次循环时(或者是其他线程enq时)，tail不为null，进入else区域。将当前线程的Node结点(简称CNode)的prev指向tail，然后使用CAS将tail指向CNode。（真实的入队情况）

acquireQueued(addWaiter(Node.EXCLUSIVE), 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);
            }
        }

可以看到，acquireQueued方法也是一个死循环，直到进入 if (p == head && tryAcquire(arg))条件方法块。

**如果当前节点的前一个节点不是头节点，就无需循环抢锁。**

**如果抢锁成功：**

1) 将CNode设置为头节点。

2) 将CNode的前置节点设置的next设置为null。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 4][]

上面操作即完成了FIFO的出队操作。

从上面的分析可以看出，只有队列的第二个节点可以有机会争用锁，如果成功获取锁，则此节点晋升为头节点。对于第三个及以后的节点，if (p == head)条件不成立，首先进行**shouldParkAfterFailedAcquire(p, node)**操作（争用锁失败的第二个节点也如此）， 来看下源码：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            int ws = pred.waitStatus;
            //如果节点的状态为-1 那就说明状态已经是可以被唤醒的状态，就直接返回true即可
            if (ws == Node.SIGNAL)
                return true;
            if (ws > 0) {
                
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
                //把前置结点变为-1，当前置结点为-1的时候我当前节点就挂起
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }

shouldParkAfterFailedAcquire方法是判断一个争用锁的线程是否应该被阻塞。它首先判断一个节点的前置节点的状态是否为Node.SIGNAL，如果是，是说明此节点已经将状态设置如果锁释放则应当通知它，所以它可以安全的阻塞了，返回true。

如果前节点的状态大于0，即为CANCELLED状态时，则会从前节点开始逐步循环找到一个没有被“CANCELLED”节点设置为当前节点的前节点，返回false。在下次循环执行shouldParkAfterFailedAcquire时，返回true。这个操作实际是把队列中CANCELLED的节点剔除掉。

前节点状态小于0的情况是对应ReentrantLock的Condition条件等待的，这里不进行展开。

private final boolean parkAndCheckInterrupt() {
            //线程挂起
            LockSupport.park(this);
            //返回一个中断标志是否被中断过，并且复位，为了响应中断
            return Thread.interrupted();
        }

如果shouldParkAfterFailedAcquire返回了true，则会执行：“parkAndCheckInterrupt()”方法，它是通过LockSupport.park(this)将当前线程挂起到WATING状态，它需要等**待一个中断、unpark方**法来唤醒它，通过这样一种FIFO的机制的等待，来实现了Lock的操作。

**获取锁的时序图：**

![image.png][image.png 1]

### 释放锁并且唤醒一下一个处于part挂起状态的线程： ###

public void unlock() {
            sync.release(1);
        }

unlock调用AQS的release()来完成, AQS的如果tryRelease方法由具体子类实现。tryRelease返回true,则会将head传入到unparkSuccessor(Node)方法中并返回true，否则返回false。

public final boolean release(int arg) {
            //如果成功
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    //唤醒自己的后继节点
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

protected final boolean tryRelease(int releases) {
                //状态值减1
                int c = getState() - releases;
                //如果当前线程不等于独占线程
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
                    free = true;
                    //设置当前独占线程为空
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }

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;
            }
            //如果有后继节点，就通过unpark来释放被挂起的线程
            if (s != null)
                LockSupport.unpark(s.thread);
        }

#  #

内部首先会发生的动作是获取head节点的next节点，如果获取到的节点不为空，则直接通过：“LockSupport.unpark()”方法来释放对应的被挂起的线程，这样一来将会有一个节点唤醒后继续进入循环进一步尝试tryAcquire()方法来获取锁。

以上ReentrantLock的释放锁的过程就分析完毕了。

[image.png]: /images/20221004/83f0b18eaa7d44f5bab80a0f6310ff19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70]: /images/20221004/e6400dcff51e42f58b92c71c3fa07ea2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 1]: /images/20221004/0bff4ff73f8e48979f6cf96cae517f4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 2]: /images/20221004/c0a2bcd368684af1882df770cba2fa71.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 3]: /images/20221004/bd706752d5a94fe18d8f8b1eeaa22803.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMwNjMxMDYz_size_16_color_FFFFFF_t_70 4]: /images/20221004/d6cbc58773e643acb3185b55c783730c.png
[image.png 1]: /images/20221004/7c7fa722b9534791b50cd90393d74891.png