做网站ssl证书必须要吗,男生十大好就业专业,网站建设公司选择意见书,杭州知名设计公司排名日常开发中#xff0c;我们经常使用锁或者其他同步器来控制并发#xff0c;那么它们的基础框架是什么呢#xff1f;如何实现的同步功能呢#xff1f;本文将详细讲解构建锁和同步器的基础框架--AQS#xff0c;并根据源码分析其原理。 一、什么是 AQS#xff1f;
(一) AQS… 日常开发中我们经常使用锁或者其他同步器来控制并发那么它们的基础框架是什么呢如何实现的同步功能呢本文将详细讲解构建锁和同步器的基础框架--AQS并根据源码分析其原理。 一、什么是 AQS
(一) AQS 简介
AQSAbstract Queued Synchronizer抽象队列同步器它是用来构建锁或其他同步器的基础框架。虽然大多数程序员可能永远不会使用到它但是知道 AQS 的原理有助于理解一些锁或同步器的是如何运行的。 那么有哪些同步器是基于 AQS 实现的呢这里仅是简单介绍详情后续会单独总结一篇文章。 同步器 说明 CountDownLatch 递减的计数器直至所有线程的任务都执行完毕才继续执行后续任务。 Semaphore 信号量控制同时访问某个资源的数量。 CyclicBarrier 递增的计数器所有线程达到屏障时才会继续执行后续任务。 ReentrantLock 防止多个线程同时访问共享资源类似 synchronized 关键字。 ReentrantReadWriteLock 维护了读锁和写锁读锁允许多线程访问读锁阻塞所有线程。 Condition 提供类似 Object 监视器的方法于 Lock 配合可以实现等待/通知模式。 FutureTask 当一个线程需要等待另一线程把某个任务执行完后才能继续执行此时可以使用 FutureTask 如果你理解了 AQS 的原理也可以基于它去自定义一个同步组件下文会介绍。
(二) AQS 数据结构
AQS 核心是通过对同步状态的管理来完成线程同步底层是依赖一个双端队列来完成同步状态的管理。
当前线程获取同步状态失败后会构造成一个 Node 节点并加入队列末尾同实阻塞线程。当同步状态释放时会把头节点中的线程唤醒让其再次尝试获取同步状态 如下图这里只是简单绘制具体流程见下面原理分析 这里的每个 Node 节点都存储着当前线程、等待信息等。
(三) 资源共享模式
我们在获取共享资源时有两种模式 模式 说明 示例 独占模式 Exclusive资源同一时刻只能被一个线程获取 ReentrantLock 共享模式 Share资源可同时被多个线程获取 Semaphore、CountDownLatch
二、AQS 原理分析
先简单说下原理分析的流程
同步状态相关源码须重写的方法Node 节点结构分析独占模式下的同步状态的获取与释放共享模式下的同步状态的获取与释放
(一) 同步状态相关
上面介绍到 AQS 核心是通过对同步状态的管理来完成线程同步所以首先介绍管理同步状态的三个方法在自定义同步组件时需要通过它们获取和修改同步状态。
//保证可见性
private volatile int state//获取当前同步状态。
protected final int getState() {return state;
}//设置当前同步状态。
protected final void setState(int newState) {state newState;
}//使用 CAS 设置当前状态保证原子性。
protected final boolean compareAndSetState(int expect, int update) {// See below for intrinsics setup to support thisreturn unsafe.compareAndSwapInt(this, stateOffset, expect, update);
} (二) 须重写的方法
AQS 是基于模板方法模式的通过第一个 abstract 也可知道AQS 是个抽象类使用者需要继承 AQS 并重写指定方法。 以下这些方式是没有具体实现的需要在使用 AQS 时在子类中去实现具体方法等到介绍一些同步组件时会详细说明如何重写。
//独占式获取同步状态实现该方法须查询并判断当前状态是否符合预期然后再进行CAS设置状态。
protected boolean tryAcquire (int arg) //独占式释放同步状态等待获取同步状态的线程将有机会获取同步状态。
protected boolean tryRelease (int arg) //共享式获取同步状态返回大于0的值表示获取成功反之获取失败。
protected int tryAcquireShared (int arg)//共享式释放同步状态。
protected boolean tryReleaseShared (int arg)//当前同步器是否再独占模式下被线程占用一般用来表示是否被当前线程独占。
protected boolean isHeldExclusively ()
(三) Node 源码
Node 是双端队列中的节点是数据结构的重要部分线程相关的信息都存在每一个 Node 中。
1. Node 结构源码
源码如下
static final class Node {//标记当前节点的线程在共享模式下等待。static final Node SHARED new Node();//标记当前节点的线程在独占模式下等待。static final Node EXCLUSIVE null;//waitStatus的值表示当前节点的线程已取消等待超时或被中断static final int CANCELLED 1;//waitStatus的值表示后继节点的线程需要被唤醒static final int SIGNAL -1;//waitStatus的值表示当前节点在等待某个条件正处于condition等待队列中static final int CONDITION -2;//waitStatus的值表示在当前有资源可用能够执行后续的acquireShared操作static final int PROPAGATE -3;//等待状态值如上1、-1、-2、-3。volatile int waitStatus;//前趋节点volatile Node prev;//后继节点volatile Node next;//当前线程volatile Thread thread;//等待队列中的后继节点共享模式下值为SHARED常量Node nextWaiter;//判断共享模式的方法final boolean isShared() {return nextWaiter SHARED;}//返回前趋节点没有报NPEfinal Node predecessor() throws NullPointerException {Node p prev;if (p null)throw new NullPointerException();elsereturn p;}//下面是三个构造方法Node() {} // Used to establish initial head or SHARED markeNode(Thread thread, Node mode) { // Used by addWaiterthis.nextWaiter mode;this.thread thread;}Node(Thread thread, int waitStatus) { // Used by Conditionthis.waitStatus waitStatus;this.thread thread;}
}
2. 设置头尾节点
Unsafe 类中提供了一个基于 CAS 的设置头尾节点的方法AQS 调用该方法进行设置头尾节点保证并发编程中的线程安全。
//CAS自旋设置头节点
private final boolean compareAndSetHead(Node update) {return unsafe.compareAndSwapObject(this, headOffset, null, update);
}//CAS自旋设置尾节点expect为当前线程“认为”的尾节点update为当前节点
private final boolean compareAndSetTail(Node expect, Node update) {return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
(四) 独占模式
资源同一时刻只能被一个线程获取如 ReentrantLock。
1. 获取同步状态
代码如下调用 acquire 方法可以获取同步状态底层就是调用须重写方法中的 tryAcquire。如果获取失败则进入同步队列中即使后续对线程进行终端操作线程也不会从同步队列中移除。 public final void acquire(int arg) {//调用须重写方法中的tryAcquireif (!tryAcquire(arg) //失败则进入同步队列中acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();
} 获取失败会先调用 addWaiter 方法将当前线程封装成独占式模式的节点添加到AQS的队列尾部源码如下。
private Node addWaiter(Node mode) {//将当前线程封装成对应模式下的Node节点Node node new Node(Thread.currentThread(), mode);Node pred tail;//尾节点if (pred ! null) {//双端队列需要两个指针指向node.prev pred;//通过CAS方式if (compareAndSetTail(pred, node)) {//添加到队列尾部pred.next node;return node;}}//等待队列中没有节点或者添加队列尾部失败则调用end方法enq(node);return node;
}//Node节点通过CAS自旋的方式被添加到队列尾部直到添加成功为止。
private Node enq(final Node node) {//死循环类似 while(1)for (;;) {Node t tail;if (t null) { // 须要初始化代表队列的第一个元素if (compareAndSetHead(new Node()))//头节点就是尾节点tail head;} else {//双端队列需要两个指针指向node.prev t;//通过自旋放入队列尾部if (compareAndSetTail(t, node)) {t.next node;return t;}}}
} 此时通过 addWaiter 已经将当前线程封装成独占模式的 Node 节点并成功放入队列尾部。接下来会调用acquireQueued 方法在等待队列中排队。
final boolean acquireQueued(final Node node, int arg) {//获取资源失败标识boolean failed true;try {//线程是否被中断标识boolean interrupted false;//死循环类似 while(1)for (;;) {//获取当前节点的前趋节点final Node p node.predecessor();//前趋节点是head即队列的第二个节点可以尝试获取资源if (p head tryAcquire(arg)) {//资源获取成功将当前节点设置为头节点setHead(node);p.next null; // help GC表示head节点出队列failed false;return interrupted;}//判断当前线程是否可以进入waitting状态详解见下方if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) //阻塞当前线程详解见下方interrupted true;}} finally {if (failed)//取消获取同步状态源码见下方的取消获取同步状态章节cancelAcquire(node);}
}//将当前节点设置为头节点
private void setHead(Node node) {head node;node.thread null;node.prev null;
}//判断当前线程是否可以进入waitting状态
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {//获取前趋节点的等待状态含义见上方Node结构源码int ws pred.waitStatus;if (ws Node.SIGNAL) //表示当前节点的线程需要被唤醒return true;if (ws 0) { //表示当前节点的线程被取消//则当前节点一直向前移动直到找到一个waitStatus状态小于或等于0的节点do {node.prev pred pred.prev;} while (pred.waitStatus 0);//排在这个节点的后面pred.next node;} else {//通过CAS设置等待状态compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;
}//阻塞当前线程
private final boolean parkAndCheckInterrupt() {//底层调用的UnSafe类的方法 park阻塞当前线程, unpark使给定的线程停止阻塞LockSupport.park(this);//中断线程return Thread.interrupted();
} acquireQueued 方法中只有当前驱节点等于 head 节点时才能够尝试获取同步状态这时为什么呢 因为 head 节点是占有资源的节点它释放后才会唤醒它的后继节点所以需要检测。还有一个原因是因为如果遇到了非 head 节点的其他节点出队或因中断而从等待中唤醒这时种情况则需要判断前趋节点是否为 head 节点是才允许获取同步状态。 获取同步状态的整体流程图如下 2. 释放同步状态
调用须重写方法中的 tryAcquire 进行同步状态的释放成功则唤醒队列中最前面的线程具体如下。
public final boolean release(int arg) {//调用须重写方法中的tryReleaseif (tryRelease(arg)) {Node h head;if (h ! null h.waitStatus ! 0)//唤醒后继节点的线程详情见下方unparkSuccessor(h);return true;}return false;
}//唤醒后继节点的线程
private void unparkSuccessor(Node node) {//获取当前节点的等待状态int ws node.waitStatus;if (ws 0)//小于0则则尝试CAS设为0compareAndSetWaitStatus(node, ws, 0);//获取后继节点Node s node.next;//后继节点为空或者等待状态大于0代表被节点被取消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;}//不为空则进行唤醒操作if (s ! null)//底层调用的UnSafe类的方法 park阻塞当前线程, unpark使给定的线程停止阻塞LockSupport.unpark(s.thread);
}
3. 其他情况的获取同步状态
除此之外独占模式下 AQS 还提供了两个获取同步状态的方法可中断的获取同步状态和超时获取同步状态。 acquire 方法获取锁失败的线程是不能被 interrupt 方法中断的所以提供了另一个方法 从而让获取锁失败等待的线程可以被中断。底层源码与
public final void acquireInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())//中断则抛出异常throw new InterruptedException();if (!tryAcquire(arg))doAcquireInterruptibly(arg);
} 通过调用 tryAcquireNanos 可以在超时时间内获取同步状态可以理解为是上述中断获取同步状态的增强版。 public final boolean tryAcquireNanos(int arg, long nanosTimeout)throws InterruptedException {if (Thread.interrupted())//中断则抛出异常throw new InterruptedException();return tryAcquire(arg) ||doAcquireNanos(arg, nanosTimeout);
} 上面个两个方法的源码均与普通的独占获取同步状态的源码基本类似感兴趣的话可以自行阅读这里不做赘述。
(五) 共享模式
资源可同时被多个线程获取如 Semaphore、CountDownLatch。
1. 获取同步状态
代码如下调用 acquireShared 方法可以获取同步状态底层就是先调用须重写方法中的 tryAcquireShared。 tryAcquireShared 返回值的含义
负数表示获取资源失败0表示获取资源成功但是没有剩余资源正数表示获取资源成功还有剩余资源 public final void acquireShared(int arg) {//调用须重写方法中的tryAcquireSharedif (tryAcquireShared(arg) 0)//获取资源失败将当前线程放入队列的尾部并阻塞doAcquireShared(arg);
} 若获取资源失败调用如下方法将当前线程放入队列的尾部并阻塞直到有其他线程释放资源并唤醒当前线程。
//部分方法与独占模式下的方法公用这里不再重复说明详情见独占模式下的获取同步状态源码。
private void doAcquireShared(int arg) {//将当前线程封装成独占式模式的节点添加到AQS的队列尾部源码在独占模式中已分析。final Node node addWaiter(Node.SHARED);//获取资源失败标识boolean failed true;try {//线程被打断表示boolean interrupted false;//死循环类似 while(1)for (;;) {//获取当前节点的前趋节点final Node p node.predecessor();//前趋节点是head即队列的第二个节点可以尝试获取资源if (p head) {int r tryAcquireShared(arg);if (r 0) {//将当前节点设置为头节点若还有剩余资源则继续唤醒队列中后面的线程。setHeadAndPropagate(node, r);p.next null; // help GC 表示head节点出队列if (interrupted)selfInterrupt();failed false;return;}}//判断当前线程是否可以进入waitting状态源码在独占模式中已分析。if (shouldParkAfterFailedAcquire(p, node) //阻塞当前线程源码在独占模式中已分析。parkAndCheckInterrupt()) interrupted true;}} finally {if (failed)//取消获取同步状态源码见下方的取消获取同步状态章节cancelAcquire(node);}
}/** propagate就是tryAcquireShared的返回值* ● 负数表示获取资源失败* ● 0表示获取资源成功但是没有剩余资源* ● 正数表示获取资源成功还有剩余资源*/
private void setHeadAndPropagate(Node node, int propagate) {//将当前节点设置为头节点源码在独占模式中已分析。Node h head; //这时的h是旧的headsetHead(node);// propagate 0还有剩余资源// h null 和 h head) null: 不会成立因为addWaiter已执行// waitStatus 0若没有剩余资源但waitStatus又小于0表示可能有新资源释放// 括号中的 waitStatus 0: 这里的 h 是此时的新的head当前节点if (propagate 0 || h null || h.waitStatus 0 ||(h head) null || h.waitStatus 0) {//获取当前节点的后继节点Node s node.next;//后继节点不存在或者是共享锁都需要唤醒可理解为只要后继节点不是独占模式都要唤醒//可能会导致不必要的唤醒if (s null || s.isShared())//唤醒操作在此方法中详情见下方的释放源码doReleaseShared();}
}
2. 释放同步状态
代码如下调用 releaseShared 方法可以释放同步状态底层就是先调用须重写方法中的 tryReleaseShared。
public final boolean releaseShared(int arg) {调用须重写方法中的tryReleaseSharedif (tryReleaseShared(arg)) {//尝试释放资源成功会继续唤醒队列中后面的线程。doReleaseShared();return true;}return false;
}//唤醒队列中后面的线程
private void doReleaseShared() {//死循环自旋操作for (;;) {//获取头节点Node h head;if (h ! null h ! tail) {int ws h.waitStatus;//signal表示后继节点需要被唤醒if (ws Node.SIGNAL) {//自旋将头节点的waitStatus状态设置为0if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue; // loop to recheck cases//唤醒头节点的后继节点源码见独占模式的释放unparkSuccessor(h);}//后继节点不需要唤醒则把当前节点状态设置为PROPAGATE确保以后可以传递下去else if (ws 0 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue; // loop on failed CAS}//判断头节点是否变化没有则退出循环。//有变化说明其他线程已经获取了同步状态需要进行重试操作。if (h head) // loop if head changedbreak;}
}
(六) 取消获取同步状态
无论是独占模式还是共享模式所有的程获取同步状态的过程中如果发生异常或是超时唤醒等都需要将当前的节点出队源码如下。 //一般在获取同步状态方法的finally块中
private void cancelAcquire(Node node) {if (node null)return;node.thread null; //当前线程节点设为nullNode pred node.prev; //获取前驱节点//前趋节点为取消状态向前遍历找到非取消状态的节点while (pred.waitStatus 0)node.prev pred pred.prev;Node predNext pred.next; //获取非取消节点的下一个节点node.waitStatus Node.CANCELLED; //将当前节点的等待状态设为取消状态//当前节点是尾节点则自旋将尾节点设置为前一个非取消节点if (node tail compareAndSetTail(node, pred)) {//将尾节点设为前一个非取消的节点并将其后继节点设为nullhelp GCcompareAndSetNext(pred, predNext, null);} else {int ws;//用于表示等待状态//pred ! head前一个非取消的节点非头节点也非尾节点//ws Node.SIGNAL当前等待状态为待唤醒//若不是待唤醒则CAS设置为待唤醒状态//前一个非取消的节点的线程不为nullif (pred ! head ((ws pred.waitStatus) Node.SIGNAL ||(ws 0 compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) pred.thread ! null) {//符合所有条件后获取当前节点的后继节点Node next node.next;if (next ! null next.waitStatus 0)//前一个非取消的节点的后继节点设为当前节点的后继节点//这样当前节点以及之前的已取消节点都会被移除compareAndSetNext(pred, predNext, next);} else {//前一个非取消的节点为头节点//唤醒后继节点的线程详情见独占模式释放同步状态的源码//唤醒是为了执行shouldParkAfterFailedAcquire()方法详解见上面的acquireQueued源码//该方法中从后往前遍历找到第一个非取消的节点并将中间的移除队列unparkSuccessor(node);}//移除当前节点node.next node; // help GC}
} 三、总结
AQS 是用来构建锁或其他同步器的基础框架底层是一个双端队列。支持独占和共享两种模式下的资源获取与释放基于 AQS 可以自定义不同类型的同步组件。 在独占模式下获取同步状态时AQS 维护了一个双端队列获取失败的线程都会被加入到队列中进行自旋移出队列的条件就是前趋节点为 head 节点并成功获取同步状态。释放同步状态时会唤醒 head 节点的后继节点。 在共享模式下获取同步状态时同样维护了一个双端队列获取失败的的线程也会加入到队列中进行自旋移除队列的条件也与独占模式一样。 但是在唤醒操作上在资源数量足够的情况下共享模式会将唤醒事件传递到后面的共享节点上进行了后续节点的唤醒解所成功后仍会唤醒后续节点。 关于 AQS 重要的几个组件的特点、原理以及对应的应用场景后续会单独写一篇文章。若发现其他问题欢迎指正交流。 参考 [1] 翟陆续/薛宾田. Java 并发编程之美. [2] 方腾飞/魏鹏/程晓明. Java 并发编程的艺术. [3] Lev Vygotsky. Java 并发编程实践
