AQS为什么要学？

• AQS是jdk并发包java.util.concurrent下绝大部分工具类实现的基础，非常重要！
• AQS是工作中并发编程常用的类Lock, Latch, Barrier等，都是基于AQS来实现的！
• 了解AQS，后面学习一些并发工具类，事半功倍！
• AQS是管程模型在java层面的实现！（jvm层面的实现是synchronized）

java层面如何实现管程？

• synchronized是自动的加锁解锁，java层面需要手动的加锁解锁。
• 并发可能有多种场景，但是有一些是相同的，可以把这些抽象出来:AQS。

AQS是什么？

• AbstractQueuedSynchronizer类，简称AQS。
• 抽象同步框架，可以用来实现一个依赖状态的同步器。
• java.util.concurrent包中的大多数同步器实现都是围绕着共同的基础行为，比如等待队列、条件队列、独占获取、共享获取…
• 适用于jdk1.5以上！

AQS的同步等待队列：

• 主要用于维护获取锁失败时入队的线程。
• 使用双向链表实现！
• 线程的使用权可以去实现一个volatile int state，只有一个线程cas 0到1成功后进行执行代码,cas 1到0成功后其他线程开始争抢使用权。
• 他是一个队列，所以需要有入队、出队操作。
• 这个锁有可能是独占锁、也可能是共享锁，所以可以将加锁、解锁的实现交给子类。

AQS的条件等待队列：

• 调用await()的时候会释放锁，然后线程会加入到条件队列，调用signal()唤醒的时候会把条件队列中的线程节点移动到同步队列中，等待再次获得锁。
• 使用单项链表实现！
• 需要提供等待唤醒机制：对比synchronized的wait/notify,notifyAll,我们也需要提供等待唤醒的方法await/signal,signalAll。
• 他是一个队列，所以需要有入队、出队操作。

AQS必须具备的功能

• 阻塞等待队列
• 共享/独占
• 公平/非公平
• 可重入
• 允许中断

AQS的内部属性：volatile int state

    /*** The synchronization state.*/private volatile int state;/*** Returns the current value of synchronization state.* This operation has memory semantics of a {@code volatile} read.* @return current state value*/protected final int getState() {return state;}/*** Sets the value of synchronization state.* This operation has memory semantics of a {@code volatile} write.* @param newState the new state value*/protected final void setState(int newState) {state = newState;}/*** Atomically sets synchronization state to the given updated* value if the current state value equals the expected value.* This operation has memory semantics of a {@code volatile} read* and write.** @param expect the expected value* @param update the new value* @return {@code true} if successful. False return indicates that the actual*         value was not equal to the expected value.*/// 核心方法，通过cas去改变State的值！protected final boolean compareAndSetState(int expect, int update) {// See below for intrinsics setup to support thisreturn unsafe.compareAndSwapInt(this, stateOffset, expect, update);}

AQS的资源共享方式

/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;

• Exclusive-独占，只有一个线程能执行，如ReentrantLock。
• Share-共享，多个线程可以同时执行，如Semaphore/CountDownLatch。

AQS的节点状态

static final int CANCELLED =  1;
static final int SIGNAL    = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;

• 值为0，初始化状态，表示当前节点在sync队列中，等待着获取锁。
• CANCELLED，值为1，表示当前的线程被取消；
• SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark；
• CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中；
• PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行；

自定义同步器实现时重点关注什么？

    // 该线程是否正在独占资源。只有用到condition才需要去实现它。protected boolean isHeldExclusively() {throw new UnsupportedOperationException();}// 独占方式。尝试获取资源，成功则返回true，失败则返回false。protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();}// 独占方式。尝试释放资源，成功则返回true，失败则返回false。protected boolean tryRelease(int arg) {throw new UnsupportedOperationException();}// 共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。protected int tryAcquireShared(int arg) {throw new UnsupportedOperationException();}// 共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。protected boolean tryReleaseShared(int arg) {throw new UnsupportedOperationException();}

我的自定义锁

public class ZhangWeiLock extends AbstractQueuedSynchronizer {/*** 尝试获取锁*/@Overrideprotected boolean tryAcquire(int arg) {// cas把state=0变为state=1if (compareAndSetState(0, 1)) {// 设置执行的线程setExclusiveOwnerThread(Thread.currentThread());// 返回获取锁成功return true;}// 返回获取锁成功return false;}/*** 释放锁*/@Overrideprotected boolean tryRelease(int arg) {// 只有获取锁的线程才能释放锁setExclusiveOwnerThread(null);setState(0);// 返回释放锁成功return true;}public void lock() {acquire(1);}public boolean tryLock() {return tryAcquire(1);}public void unlock() {release(1);}public boolean isLocked() {return isHeldExclusively();}}

测试方法

public class TsetZhangWeiLock {// 定义数据private static int sum = 0;// 定义自己的锁private static ZhangWeiLock lock = new ZhangWeiLock();public static void main(String[] args) throws InterruptedException {// 循环3吃，开启三个线程for (int i = 0; i < 3; i++) {Thread thread = new Thread(() -> {// 加锁lock.lock();try {// 不加锁这里执行完小于30000for (int j = 0; j < 10000; j++) {sum++;}} finally {// 解锁lock.unlock();}});thread.start();}// 等待线程执行完Thread.sleep(2000);// 打印最后的结果System.out.println(sum);}
}

测试结果

JDK中是如何自定义同步器的？

• 一般是通过一个内部类Sync继承 AQS
• 将同步器所有调用都映射到Sync对应的方法
• 后续基于源码深入分析各种场景的锁。

AQS的等待唤醒机制内部类：Condition

• 调用Condition#await方法会释放当前持有的锁，然后阻塞当前线程，同时向Condition队列尾部添加一个节点，所以调用Condition#await方法的时候必须持有锁。
• 调用Condition#signal方法会将Condition队列的首节点移动到阻塞队列尾部，然后唤醒因调用Condition#await方法而阻塞的线程(唤醒之后这个线程就可以去竞争锁了)，所以调用Condition#signal方法的时候必须持有锁，持有锁的线程唤醒被因调用Condition#await方法而阻塞的线程。

在AbstractQueuedSynchronizer类中的源码位置

    public class ConditionObject implements Condition, java.io.Serializable {

Condition接口的方法

验证等待唤醒：使用ReentrantLock（后续有针对ReentrantLock深入的源码分析）

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;import lombok.extern.slf4j.Slf4j;@Slf4j
public class TsstWait {public static void main(String[] args) {Lock lock = new ReentrantLock();Condition condition = lock.newCondition();new Thread(() -> {lock.lock();try {log.debug(Thread.currentThread().getName() + " 开始处理任务");condition.await();log.debug(Thread.currentThread().getName() + " 结束处理任务");} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}).start();new Thread(() -> {lock.lock();try {log.debug(Thread.currentThread().getName() + " 开始处理任务");Thread.sleep(2000);condition.signal();log.debug(Thread.currentThread().getName() + " 结束处理任务");} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}).start();}
}

等待唤醒结果

结束语

• 获取更多有价值的文章，让我们一起成为架构师！
• 关注公众号，可以让你对MySQL有非常深入的了解
• 关注公众号，每天持续高效的了解并发编程！
• 关注公众号，后续持续高效的了解spring源码！
• 这个公众号，无广告！！！每日更新！！！