Java并发技术总结之五——AQS

剑影啸清寒

发布于 2020-07-15 15:56:09

3510

发布于 2020-07-15 15:56:09

文章被收录于专栏：琦小虾的Binary

接上篇《Java并发技术总结之四——CAS》

五. AQS 原理

参考地址：《Java并发-AQS及各种Lock锁的原理》《JAVA并发编程: CAS和AQS》《Java并发之AQS详解》

5.1 AQS 简介

AQS (AbustactQueuedSynchronizer) 是 Java 提供的底层同步工具类，主要思想是用一个 int 类型的变量表示同步状态，以及一个双链表形式的同步队列，并提供了一系列的 CAS (Compare And Swap) 操作来管理这个同步状态。

AQS 的主要作用是为 Java 中的并发同步组件提供统一的底层支持，例如 ReentrantLock，CountDownLatch 就是基于 AQS 实现的，实现方法是通过继承 AQS 实现其模版方法，然后将子类作为同步组件的内部类。

5.2 AQS 基本方法

AQS 有若干基本方法：

boolean tryAcquire(int arg);
boolean tryRelease(int arg);
int tryAcquireShared(int arg);
boolean tryReleaseShared(int arg);
boolean isHeldExclusively();

以上方法不需要全部实现，根据获取的锁的种类可以选择实现不同的方法。支持独占（排他）锁的同步器，比如 ReentrantLock，应该实现 tryAcquire, tryRelease, isHeldExclusively 方法；而作为共享锁的同步器，比如 CountDownLatch，应该实现 tryAcquireShared, tryReleaseShared 方法。

5.3 同步队列

同步队列是 AQS 很重要的组成部分，它是一个双端队列，遵循 FIFO 原则，主要作用是存放在锁上阻塞的线程。比如可重入锁 ReentrantLock 的 lock(), unlock() 方法，分别实现了线程挂起、释放，本质上就是将线程存入同步队列、弹出同步队列的操作。

5.3.1 独占锁 - 获取锁

对于独占锁 (如 ReentrantLock)，需要实现 tryAcquire, tryRelease, isHeldExclusively 方法。当一个线程尝试获取锁时，如果已经被占用，那么当前线程就会被构造成一个 Node 节点，加入到同步队列的尾部。如下图所示：

线程尝试获取锁的操作，本质上就是将当前线程加入到同步队列尾部的操作。步骤如下：

调用 AQS 的入口方法 acquire(int arg)；
调用 AQS 的模板方法 tryAcquire(int arg) 尝试获取锁；如果获取锁成功，则不进入同步队列；
如果尝试获取锁失败，则将当前线程构造成一个 Node 节点，通过 CAS 将其加入到同步队列尾部，然后该 Node 对应的线程进入自旋状态；
- 自旋状态下，判断两个条件：
  - 同步队列的前驱节点 prev 是否为头结点 head；
  - tryAcquire() 是否获取成功；
- 两个条件同时成立，则将当前线程节点设置为头结点；
- 不同时成立，使用 LockSupport.park(this) 方法将当前线程挂起，等待 prev 节点的唤醒；

8.3.2 独占锁 - 释放锁

队列的头节点是成功获取锁的节点，对于独占锁 (如 ReentrantLock)，当头节点线程释放锁时，会唤醒后面的节点并释放当前头节点的引用。如下图所示：

线程释放锁的操作，本质上就是将当前线程从同步队列头部弹出的操作。步骤如下：

调用 AQS 的入口方法 release(int arg)；
调用 AQS 的模板方法 tryRelease(int arg) 尝试释放同步状态；
将当前节点的下一个节点作为头结点 head；
通过 LockSupport.unpark(currentNode.next.thread) 唤醒后继节点（即线程入队的步骤 3）；

5.3.3 共享锁 - 获取锁

CountDownLatch, CyclicBarrier, Semaphore 都属于共享锁的同步器，应该实现 tryAcquireShared, tryReleaseShared 方法。获取锁流程如下：

调用 AQS 的入口方法 acquireShared(int arg)；
进入 AQS 的模板方法 tryAcquireShared(int arg)，获取同步状态；
- 返回值 >= 0，则说明同步状态有剩余，获取锁成功，直接返回；
- 返回值 < 0，则说明获取同步锁失败，向队尾添加共享类型的 Node 节点 (Node.SHARED)，该 Node 对应的线程进入自旋状态；
自旋状态下，判断两个条件：
- 条件 1：同步队列的前驱节点 prev 是否为头结点 head；
- 条件 2：tryAcquire() 是否获取成功；
- 两个条件同时成立，则将当前线程节点设置为头结点，并唤醒所有后继节点，所有后继节点重新进入尝试获取锁的状态；
- 不同时成立，使用 LockSupport.park(this) 方法将当前线程挂起，等待 prev 节点的唤醒；

需要注意的是，自旋状态下独占锁与共享锁判断的条件相同，但执行动作不同：独占锁只将当前节点设置为头结点，共享锁在此之外还唤醒所有的后继节点。

5.3.4 共享锁 - 释放锁

调用 releaseShared(int arg) 模板方法释放同步状态；
如果释放成功，则遍历整个队列，使用 LockSupport.unpart(nextNode.thread) 唤醒所有后继节点；

5.3.5 独占锁与共享锁的区别

同步状态值
- 独占锁的同步状态值 state = 1，即同一时刻只能有一个线程成功获取同步状态；
- 共享锁的同步状态 state > 1，取值由上层同步组件确定；此外，共享锁会出现多个线程同时成功获取同步状态的情况；
头结点运行后操作
- 独占锁的同步队列中，Node 节点运行完毕后，释放该节点后面一个后继节点（即直接后继节点）；
- 共享锁的同步队列中，Node 节点运行完毕后，释放该节点后面所有后继节点；

5.4 可重入锁 (ReentrantLock) 实现原理

重入锁指的是当前线程成功获取锁后，如果再次访问该临界区，则不会对自己产生互斥行为。Java 中的 ReentrantLock 和 synchronized 关键字都是可重入锁，synchronized 由 JVM 实现可重入性，ReentrantLock 的可重入性基于 AQS 实现。此外，ReentrantLock 还提供公平锁和非公平锁两种模式，默认创建非公平锁。

重入锁的基本原理，是判断上次获取锁的线程是否为当前线程，如果是则可再次进入临界区，如果不是，则阻塞。由于 ReentrantLock 是基于 AQS 实现的，底层通过操作同步状态来获取锁。

5.4.1 非公平锁实现

下面看一下非公平锁的实现逻辑代码如下：

    final boolean nonfairTryAcquire(int acquires) {
        // 获取当前线程
        final Thread current = Thread.currentThread();
        // 通过 AQS 获取同步状态
        int c = getState();
        // 同步状态为 0，说明临界区处于无锁状态，
        if (c == 0) {
            // 修改同步状态，即加锁
            if (compareAndSetState(0, acquires)) {
                // 将当前线程设置为锁的 owner
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        // 如果临界区处于锁定状态，且上次获取锁的线程为当前线程
        else if (current == getExclusiveOwnerThread()) {
            // 则递增同步状态
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

非公平锁是指在锁状态为可用时，所有等待该锁释放的线程有相同的权利争抢该锁，与线程等待时间长短无关。只要同步状态 state = 0 任意调用 lock() 方法，都有可能获取到锁。

注：在 compareAndSetState 方法中，使用了 CAS 的比较交换方法。关于 CAS 见其他部分的讲解。

5.4.2 公平锁实现

公平锁是指锁状态可用时，对于所有正在等待该锁释放的线程，按照等待时间进行排序，等待时间最长的线程获取该锁。公平锁与非公平锁的实现逻辑基本相同，逻辑不同的地方主要是在获取到线程状态 state = 0 时的处理。关键代码如下：

    if (c == 0) {
        // 此处为公平锁的核心，即判断同步队列中当前节点是否有前驱节点
        if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }

代码中的 hasQueuedPredecessors() 方法是关键区别，用来判断是否拥有前驱节点。对于公平锁来说，如果同步队列中拥有前驱节点，说明在该 Node 对应的线程之前，还有其他线程存入了同步队列中，那么就不满足获取该锁的条件。

5.5 读写锁 (ReadWriteReentrantLock) 实现原理

Java 提供了一个基于 AQS 到读写锁实现 ReentrantReadWriteLock，该读写锁到实现原理是：将同步变量 state 按照高 16 位和低 16 位进行拆分，读锁为高 16 位，是共享锁；写锁是低 16 位，是独占锁。如下图所示。

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-pcHP2JSQ-1594653931949)(./pic/Java读写锁划分方式.png)

5.5.1 写锁

写锁是一个独占锁，获取锁的主要逻辑在 ReentrantReadWriteLock # tryAcquire(int arg) 中实现：

    protected final boolean tryAcquire(int acquires) {
        Thread current = Thread.currentThread();
        int c = getState();
        int w = exclusiveCount(c);
        if (c != 0) {
            if (w == 0 || current != getExclusiveOwnerThread())
                return false;
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // Reentrant acquire
            setState(c + acquires);
            return true;
        }
        if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
            return false;
        setExclusiveOwnerThread(current);
        return true;
    }

获取同步状态，同时从同步状态中分离出低 16 位的写锁状态；
如果同步状态不为 0，说明当前状态下存在读锁或者写锁；
如果存在读锁，那么不能获取写锁；这样是为了保证对读的可见性；
- 存在读锁的判断：c != 0 && w == 0
如果当前线程不是写锁的线程，不能获取写锁；
上面的判断全部通过，则用 CAS 将锁同步状态进行修改，最后将当前线程设置为写锁的获取线程；
- 修改同步状态，是通过修改同步状态低 16 位的写锁完成的；

写锁的释放逻辑与独占锁基本相同。代码如下：

    protected final boolean tryRelease(int releases) {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        int nextc = getState() - releases;
        boolean free = exclusiveCount(nextc) == 0;
        if (free)
            setExclusiveOwnerThread(null);
        setState(nextc);
        return free;
    }

释放过程中，不断减少读锁的同步状态，直到读锁同步状态为 0 时，写锁完全被释放。

5.5.2 读锁

读锁是一个共享锁，获取读锁的步骤如下：

    protected final int tryAcquireShared(int unused) {
        Thread current = Thread.currentThread();
        int c = getState();
        if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
            return -1;
        int r = sharedCount(c);
        if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
            if (r == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
            }
            return 1;
        }
        return fullTryAcquireShared(current);
    }

获取当前同步状态；
计算高 16 位读锁状态 r；
异常判断：如果 r + 1 大于获取到读锁的最大值，则抛出异常；
如果存在写锁，而且当前线程不是写锁的获取者，则获取读锁失败；
如果上述所有判断都通过，则通过 CAS 重新设置读锁同步状态；

读锁的释放与写锁类似，不断的释放写锁状态，直到为 0，表示没有线程持有读锁。

5.6 CountDownLatch 原理解析

CountDownLatch 本身是使用 AQS 的共享锁实现的。用法与原理如下：

5.6.1 CountDownLatch 用法

CountDownLatch 类位于 java.util.concurrent 包下，利用它可以实现类似计数器的功能。比如有一个任务 A，它要等待其他 4 个任务执行完毕之后才能执行，此时就可以利用 CountDownLatch 来实现这种功能了。

CountDownLatch 类只提供了一个构造器：

//参数count为计数值
public CountDownLatch(int count) {  }

然后下面这 3 个方法是 CountDownLatch 类中最重要的方法：

// 调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
public void await() throws InterruptedException { };
// 和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };
// 将count值减1
public void countDown() { };

下面看一个例子大家就清楚CountDownLatch的用法了：

public class Test {
     public static void main(String[] args) {   
         final CountDownLatch latch = new CountDownLatch(2);
 
         new Thread(){
             public void run() {
                 try {
                     System.out.println("子线程"+Thread.currentThread().getName()+"正在执行");
                    Thread.sleep(3000);
                    System.out.println("子线程"+Thread.currentThread().getName()+"执行完毕");
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
             };
         }.start();
 
         new Thread(){
             public void run() {
                 try {
                     System.out.println("子线程"+Thread.currentThread().getName()+"正在执行");
                     Thread.sleep(3000);
                     System.out.println("子线程"+Thread.currentThread().getName()+"执行完毕");
                     latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
             };
         }.start();
 
         try {
             System.out.println("等待2个子线程执行完毕...");
            latch.await();
            System.out.println("2个子线程已经执行完毕");
            System.out.println("继续执行主线程");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
     }
}

执行结果：

线程Thread-0正在执行
线程Thread-1正在执行
等待2个子线程执行完毕...
线程Thread-0执行完毕
线程Thread-1执行完毕
2个子线程已经执行完毕
继续执行主线程

5.6.2 CountDownLatch 原理

创建 CountDownLatch 对象时，首先设置一个 count 值，这个值对应着 AQS 的 state 值；
CountDownLatch 运行到最后，都会调用 countDownLatch.await() 方法，该操作本质是尝试获取共享锁，而且成功获取共享锁的标志是 state == 0；在 state != 0 的情况下，await() 方法被阻塞；
state 值变化的操作一般是通过 countDownLatch.countDown() 方法实现的，通常每个线程执行完自己的工作后，执行 countDown() 方法，该操作的本质是使用 CAS 将 state 值改为 state - 1；
等到正常运行结束，state == 0；await() 正常获取到锁，阻塞解除，主线程正常继续运行。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2020/07/13 ，如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法