当前位置：网站首页>AQS原理

AQS原理

2022-07-28 23:01:00 【51CTO】

AQS 原理

1. 概述

全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架。子类继承他来实现同步器。底层实现可参照Monitor理解。主要通过CAS和volitile维护state从而实现的锁。

AQS原理_java

源码展示：

AQS原理_java_02

特点：

AQS是个抽象类。AQS及其实现类，类似于Monitor。看下面第二条和第三条。但是Monitor是不可见的，是操作系统提供的。AQS是Java通过CAS和volitile实现的类似于Monitor的同步器。

用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁

getState - 获取 state 状态
setState - 设置 state 状态
compareAndSetState - cas 机制设置 state 状态
独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源

提供了基于 FIFO 的等待队列来实现阻塞队列，类似于 Monitor 的 EntryList。有一个Node内部类，用来实现FIFO的等待队列。后续讲ReentrantLock实现原理时，会具体介绍。
条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet。使用Lock.newCondition()生成条件变量，底层再用同步器（即AQS的实现类）新建。

AQS原理_java_04

AQS中的条件变量class

AQS原理_并发_05

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）

tryAcquire
tryRelease
tryAcquireShared
tryReleaseShared
isHeldExclusively

获取锁和释放锁。后续讲ReentrantLock实现原理时，会具体介绍。

获取锁的姿势

// 如果获取锁失败
if (!tryAcquire(arg)) {
// 入队, 可以选择阻塞当前线程 park unpark
}

释放锁的姿势

// 如果释放锁成功
if (tryRelease(arg)) {
// 让阻塞线程恢复运行
}

2. 实现不可重入锁

自定义同步器

AQS原理_java_02

       
       final 
       
       class 
       
       MySync 
       
       extends 
       
       AbstractQueuedSynchronizer {
       
       @Override
       
       protected 
       
       boolean 
       
       tryAcquire(
       
       int 
       
       acquires) {
       
       if (
       
       acquires 
       
       == 
       
       1){
       
       if (
       
       compareAndSetState(
       
       0, 
       
       1)) {
       
       setExclusiveOwnerThread(
       
       Thread.
       
       currentThread());
       
       return 
       
       true;
       
            }
       
        }
       
       return 
       
       false;
       
    }
       
       @Override
       
       protected 
       
       boolean 
       
       tryRelease(
       
       int 
       
       acquires) {
       
       if(
       
       acquires 
       
       == 
       
       1) {
       
       if(
       
       getState() 
       
       == 
       
       0) {
       
       throw 
       
       new 
       
       IllegalMonitorStateException();
       
            }
       
       setExclusiveOwnerThread(
       
       null);
       
       setState(
       
       0);
       
       return 
       
       true;
       
        }
       
       return 
       
       false;
       
    }
       
       protected 
       
       Condition 
       
       newCondition() {
       
       return 
       
       new 
       
       ConditionObject();
       
    }
       
       @Override
       
       protected 
       
       boolean 
       
       isHeldExclusively() {
       
       return 
       
       getState() 
       
       == 
       
       1;
       
    }}
      
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.

自定义锁

有了自定义同步器，很容易复用 AQS ，实现一个功能完备的自定义锁。

根据lock方法可以看出同步器sync相当于monitor，有类似于monitor的EntryList的等待队列。

       
       class 
       
       MyLock 
       
       implements 
       
       Lock {
       
       static 
       
       MySync 
       
       sync 
       
       = 
       
       new 
       
       MySync();
       
       @Override
       
       // 尝试，不成功，进入等待队列
       
       public 
       
       void 
       
       lock() {
       
       sync.
       
       acquire(
       
       1);    
       
       //acquire方法是AQS的方法
       
    }
       
       @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();
       
    }
       
}
      
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.

测试：

       
       MyLock 
       
       lock 
       
       = 
       
       new 
       
       MyLock();
       
       new 
       
       Thread(() 
       
       -> {
       
       lock.
       
       lock();    
       
       //获取成功，进入AQS的实现类MySync同步器成为Owner线程
       
       try {
       
       log.
       
       debug(
       
       "locking...");
       
       sleep(
       
       1);
       
            } 
       
       finally {
       
       log.
       
       debug(
       
       "unlocking...");
       
       lock.
       
       unlock();
       
            }
       
        },
       
       "t1").
       
       start();
       
       new 
       
       Thread(() 
       
       -> {
       
       lock.
       
       lock();        
       
       //获取失败，MySync创建阻塞队列，将t2线程入队列。
       
       try {
       
       log.
       
       debug(
       
       "locking...");
       
            } 
       
       finally {
       
       log.
       
       debug(
       
       "unlocking...");
       
       lock.
       
       unlock();
       
            }
       
        },
       
       "t2").
       
       start();
      
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.

NonfairSync继承自AQS。先简单看一下，后面会详细介绍。

AQS原理_jvm_11

输出：

22:29:28.727 c.TestAqs [t1] - locking...
22:29:29.732 c.TestAqs [t1] - unlocking...
22:29:29.732 c.TestAqs [t2] - locking...
22:29:29.732 c.TestAqs [t2] - unlocking...

不可重入测试
如果改为下面代码，会发现自己也会被挡住（只会打印一次 locking）

lock.lock();
log.debug("locking...");
lock.lock();
log.debug("locking...");

3. 心得

起源

早期程序员会自己通过一种同步器去实现另一种相近的同步器，例如用可重入锁去实现信号量，或反之。这显然不够优雅，于是在 JSR166（java 规范提案）中创建了 AQS，提供了这种通用的同步器机制。

目标

AQS 要实现的功能目标

阻塞版本获取锁 acquire 和非阻塞的版本尝试获取锁 tryAcquire(尝试失败不进入等待队列)
获取锁超时机制
通过打断取消机制
独占机制及共享机制
条件不满足时的等待机制

要实现的性能目标

Instead, the primary performance goal here is scalability: to predictably maintain efficiency even, or especially, when synchronizers are contended
相反，这里的主要性能目标是可伸缩性:在争用同步器时，可以预见地保持效率。

设计

AQS 的基本思想其实很简单

获取锁的逻辑

while(state 状态不允许获取) {
if(队列中还没有此线程) {
入队并阻塞
}
}
当前线程出队

释放锁的逻辑

if(state 状态允许了) {
恢复阻塞的线程(s)
}
unpark表示唤醒后面的一个线程。后边会详细介绍。

要点

        ● 原子维护 state 状态（通过CAS和volitile）
        ● 阻塞及恢复线程
        ● 维护队列

1) state 设计

state 使用 volatile 配合 cas 保证其修改时的原子性。
state 使用了 32bit int 来维护同步状态，因为当时使用 long 在很多平台下测试的结果并不理想。

2) 阻塞恢复设计

早期的控制线程暂停和恢复的 api 有 suspend 和 resume，但它们是不可用的，因为如果先调用的 resume，那么 suspend 将感知不到
解决方法是使用 park & unpark 来实现线程的暂停和恢复，具体原理在之前讲过了，先 unpark 再 park 也没问题
park & unpark 是针对线程的，而不是针对同步器的，因此控制粒度更为精细
park 线程还可以通过 interrupt 打断

3) 队列设计

使用了 FIFO 先入先出队列，来实现等待队列（类似于Monitor的EntryList）和条件变量。并不支持优先级队列。
设计时借鉴了 CLH（同步队列）队列，它是一种单向无锁队列。

队列中有 head 和 tail 两个指针节点，都用 volatile 修饰配合 cas 使用，每个Node有 state 维护节点状态。

入队和出队之后会详细介绍。
入队伪代码，只需要考虑 tail 赋值的原子性。

AQS原理_java_15

出队伪代码

AQS原理_ide_16

CLH 好处：

无锁，使用自旋
快速，无阻塞

AQS 在一些方面改进了 CLH

       
       private 
       
       Node 
       
       enq(
       
       final 
       
       Node 
       
       node) {
       
       for (;;) {
       
       Node 
       
       t 
       
       = 
       
       tail;
       
       // 队列中还没有元素 tail 为 null
       
       if (
       
       t 
       
       == 
       
       null) {
       
       // 将 head 从 null -> dummy
       
       if (
       
       compareAndSetHead(
       
       new 
       
       Node()))
       
       tail 
       
       = 
       
       head;
       
            } 
       
       else {
       
       // 将 node 的 prev 设置为原来的 tail
       
       node.
       
       prev 
       
       = 
       
       t;
       
       // 将 tail 从原来的 tail 设置为 node
       
       if (
       
       compareAndSetTail(
       
       t, 
       
       node)) {
       
       // 原来 tail 的 next 设置为 node
       
       t.
       
       next 
       
       = 
       
       node;
       
       return 
       
       t;
       
                }
       
            }
       
        }
       
    }
      
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.