优化型网站是什么意思,dedecms小说网站模板,wordpress不显示媒体库图片,开发项目管理工具AQS介绍AQS#xff0c;即AbstractQueuedSynchronizer, 队列同步器#xff0c;它是Java并发用来构建锁和其他同步组件的基础框架。来看下同步组件对AQS的使用#xff1a;AQS是一个抽象类#xff0c;主是是以继承的方式使用。AQS本身是没有实现任何同步接口的#xff0c;它仅… AQS介绍 AQS即AbstractQueuedSynchronizer, 队列同步器它是Java并发用来构建锁和其他同步组件的基础框架。来看下同步组件对AQS的使用 AQS是一个抽象类主是是以继承的方式使用。AQS本身是没有实现任何同步接口的它仅仅只是定义了同步状态的获取和释放的方法来供自定义的同步组件的使用。从图中可以看出在java的同步组件中AQS的子类Sync等一般是同步组件的静态内部类即通过组合的方式使用。 抽象的队列式的同步器AQS定义了一套多线程访问共享资源的同步器框架许多同步类实现都依赖于它如常用的ReentrantLock/Semaphore/CountDownLatch 它维护了一个volatile int state代表共享资源和一个FIFO双向队列线程等待队列多线程争用资源被阻塞时会进入此队列 AQS原理简介 AQS的实现依赖内部的同步队列FIFO双向队列如果当前线程获取同步状态失败AQS会将该线程以及等待状态等信息构造成一个Node将其加入同步队列的尾部同时阻塞当前线程当同步状态释放时唤醒队列的头节点。 上面说的有点抽象来具体看下首先来看AQS最主要的三个成员变量 private transient volatile Node head;private transient volatile Node tail; private volatile int state; 上面提到的同步状态就是这个int型的变量state. head和tail分别是同步队列的头结点和尾结点。假设state0表示同步状态可用如果用于锁则表示锁可用state1表示同步状态已被占用锁被占用。 下面举例说下获取和释放同步状态的过程 获取同步状态 假设线程A要获取同步状态这里想象成锁方便理解初始状态下state0,所以线程A可以顺利获取锁A获取锁后将state置为1。在A没有释放锁期间线程B也来获取锁此时因为state1表示锁被占用所以将B的线程信息和等待状态等信息构成出一个Node节点对象放入同步队列head和tail分别指向队列的头部和尾部此时队列中有一个空的Node节点作为头点head指向这个空节点空Node的后继节点是B对应的Node节点tail指向它同时阻塞线程B(这里的阻塞使用的是LockSupport.park()方法)。后续如果再有线程要获取锁都会加入队列尾部并阻塞。 释放同步状态 当线程A释放锁时即将state置为0此时A会唤醒头节点的后继节点所谓唤醒其实是调用LockSupport.unpark(B)方法即B线程从LockSupport.park()方法返回此时B发现state已经为0所以B线程可以顺利获取锁B获取锁后B的Node节点随之出队。 上面只是简单介绍了AQS获取和释放的大致过程下面结合AQS和ReentrantLock源码来具体看下JDK是如何实现的特别要注意JDK是如何保证同步和并发操作的。 AQS源码分析 接下来以ReentrantLock的源码入手来深入理解下AQS的实现。上面说过AQS一般是以继承的方式被使用同步组件内部组合一个继承了AQS的子类。在ReentrantLock类中有一个Sync成员变量即是继承了AQS的子类源码如下 public class ReentrantLock implements Lock, java.io.Serializable { private static final long serialVersionUID 7373984872572414699L; /** Synchronizer providing all implementation mechanics */ private final Sync sync; /** * Base of synchronization control for this lock. Subclassed * into fair and nonfair versions below. Uses AQS state to * represent the number of holds on the lock. */ abstract static class Sync extends AbstractQueuedSynchronizer { ... } } 这里的Sync也是一个抽象类其实现类为FairSync和NonfairSync分别对应公平锁和非公平锁。ReentrantLock的提供一个入参为boolean值的构造方法来确定使用公平锁还是非公平锁 public ReentrantLock(boolean fair) { sync fair ? new FairSync() : new NonfairSync(); } 获取锁 这里以NonfairSync类为例看下它的Lock()的实现 final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } lock方法先通过CAS尝试将同步状态(AQS的state属性)从0修改为1。若直接修改成功了则将占用锁的线程设置为当前线程。看下compareAndSetState()和setExclusiveOwnerThread()实现 protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); } 可以看到compareAndSetState底层其实是调用的unsafe的CAS系列方法。 protected final void setExclusiveOwnerThread(Thread thread) {exclusiveOwnerThread thread; } exclusiveOwnerThread属性是AQS从父类AbstractOwnableSynchronizer中继承的属性用来保存当前占用同步状态的线程。 如果CAS操作未能成功说明state已经不为0此时继续acquire(1)操作这个acquire()由AQS实现提供 public final void acquire(int arg) { if (!tryAcquire(arg) acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } 代码很短不太好了理解转换下写法(代码1) public final void acquire(int arg) { boolean hasAcquired tryAcquire(arg); if (!hasAcquired) { Node currentThreadNode addWaiter(Node.EXCLUSIVE); boolean interrupted acquireQueued(currentThreadNode, arg); if (interrupted) { selfInterrupt(); } } } 简单解释下tryAcquire方法尝试获取锁如果成功就返回如果不成功则把当前线程和等待状态信息构适成一个Node节点并将结点放入同步队列的尾部。然后为同步队列中的当前节点循环等待获取锁直到成功。 首先看tryAcquire(arg)在NonfairSync中的实现(这里arg1): protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } final boolean nonfairTryAcquire(int acquires) { final Thread current Thread.currentThread(); int c getState(); if (c 0) { if (compareAndSetState(0, acquires)) { 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; } 首先获取AQS的同步状态(state)在锁中就是锁的状态如果状态为0则尝试设置状态为arg(这里为1), 若设置成功则表示当前线程获取锁返回true。这个操作外部方法lock()就做过一次这里再做只是为了再尝试一次尽量以最简单的方式获取锁。 如果状态不为0再判断当前线程是否是锁的owner(即当前线程在之前已经获取锁这里又来获取)如果是owner, 则尝试将状态值增加acquires如果这个状态值越界抛出异常如果没有越界则设置后返回true。这里可以看非公平锁的涵义即获取锁并不会严格根据争用锁的先后顺序决定。这里的实现逻辑类似synchroized关键字的偏向锁的做法即可重入而不用进一步进行锁的竞争也解释了ReentrantLock中Reentrant的意义。 如果状态不为0且当前线程不是owner则返回false。回到上面的代码1tryAcquire返回false,接着执行addWaiter(Node.EXCLUSIVE)这个方法创建结点并入队来看下源码 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(tnull)区域从方法名可以看出这里是用CAS的方式创建一个空的Node作为头结点因为此时队列中只一个头结点所以tail也指向它第一次循环执行结束。注意这里使用CAS是防止多个线程并发执行到这儿时只有一个线程能够执行成功防止创建多个同步队列。 进行第二次循环时(或者是其他线程enq时)tail不为null进入else区域。将当前线程的Node结点(简称CNode)的prev指向tail然后使用CAS将tail指向CNode。看下这里的实现 private final boolean compareAndSetTail(Node expect, Node update) {return unsafe.compareAndSwapObject(this, tailOffset, expect, update);}expect为t, t此时指向tail,所以可以CAS成功将tail重新指向CNode。此时t为更新前的tail的值即指向空的头结点t.nextnode就将头结点的后续结点指向CNode返回头结点。经过上面的操作头结点和CNode的关系如图 其他线程再插入节点以此类推都是在追加到链表尾部并且通过CAS操作保证线程安全。 通过上面分析可知AQS的写入是一种双向链表的插入操作至此addWaiter分析完毕。 addWaiter返回了插入的节点作为acquireQueued方法的入参看下源码 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))条件方法块。还是接着刚才的操作来分析。acquireQueued接收的参数是addWaiter方法的返回值也就是刚才的CNode节点arg1。node.predecessor()返回CNode的前置节点在这里也就是head节点所以phead成立进而进行tryAcquire操作即争用锁, 如果获取成功则进入if方法体看下接下来的操作 1) 将CNode设置为头节点。2) 将CNode的前置节点设置的next设置为null。 此时队列如图 上面操作即完成了FIFO的出队操作。从上面的分析可以看出只有队列的第二个节点可以有机会争用锁如果成功获取锁则此节点晋升为头节点。对于第三个及以后的节点if (p head)条件不成立首先进行shouldParkAfterFailedAcquire(p, node)操作争用锁失败的第二个节点也如此 来看下源码 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws pred.waitStatus;if (ws Node.SIGNAL) /* * This node has already set status asking a release * to signal it, so it can safely park. */ return true; if (ws 0) { /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ do { node.prev pred pred.prev; } while (pred.waitStatus 0); pred.next node; } else { /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but dont park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } shouldParkAfterFailedAcquire方法是判断一个争用锁的线程是否应该被阻塞。它首先判断一个节点的前置节点的状态是否为Node.SIGNAL如果是是说明此节点已经将状态设置如果锁释放则应当通知它所以它可以安全的阻塞了返回true。 如果前节点的状态大于0即为CANCELLED状态时则会从前节点开始逐步循环找到一个没有被“CANCELLED”节点设置为当前节点的前节点返回false。在下次循环执行shouldParkAfterFailedAcquire时返回true。这个操作实际是把队列中CANCELLED的节点剔除掉。 前节点状态小于0的情况是对应ReentrantLock的Condition条件等待的这里不进行展开。 如果shouldParkAfterFailedAcquire返回了true则会执行“parkAndCheckInterrupt()”方法它是通过LockSupport.park(this)将当前线程挂起到WATING状态它需要等待一个中断、unpark方法来唤醒它通过这样一种FIFO的机制的等待来实现了Lock的操作。 释放锁 通过ReentrantLock的unlock方法来看下AQS的锁释放过程。来看下源码 public void unlock() { sync.release(1); } public final boolean release(int arg) { if (tryRelease(arg)) { Node h head; if (h ! null h.waitStatus ! 0) unparkSuccessor(h); return true; } return false; } unlock调用AQS的release()来完成, AQS的如果tryRelease方法由具体子类实现。tryRelease返回true,则会将head传入到unparkSuccessor(Node)方法中并返回true否则返回false。首先来看看Sync中tryRelease(int)方法实现如下所示 protected final boolean tryRelease(int releases) { 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 } 这个动作可以认为就是一个设置锁状态的操作而且是将状态减掉传入的参数值参数是1如果结果状态为0就将排它锁的Owner设置为null以使得其它的线程有机会进行执行。在排它锁中加锁的时候状态会增加1当然可以自己修改这个值在解锁的时候减掉1同一个锁在可以重入后可能会被叠加为2、3、4这些值只有unlock()的次数与lock()的次数对应才会将Owner线程设置为空而且也只有这种情况下才会返回true。 在方法unparkSuccessor(Node)中就意味着真正要释放锁了它传入的是head节点head节点是占用锁的节点看下源码 private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws node.waitStatus; if (ws 0) compareAndSetWaitStatus(node, ws, 0); /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ 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; } if (s ! null) LockSupport.unpark(s.thread); } 内部首先会发生的动作是获取head节点的next节点如果获取到的节点不为空则直接通过“LockSupport.unpark()”方法来释放对应的被挂起的线程这样一来将会有一个节点唤醒后继续进入循环进一步尝试tryAcquire()方法来获取锁。 转载于:https://www.cnblogs.com/fanBlog/p/9336126.html
