抽象同步队列AQS——AbstractQueuedSynchronizer锁详解 顶
欢迎来到阿八个人博客网站。本 阿八个人博客 网站提供最新的站长新闻,各种互联网资讯。 喜欢本站的朋友可以收藏本站,或者加QQ:我们大家一起来交流技术! URL链接:https://www.abboke.com/jsh/2019/0719/9225.html
撸了今年阿里、网易和美团的面试,我有一个重要发现.......>>>
AQS——锁的底层支持
谈到并发,不得不谈ReentrantLock;而谈到ReentrantLock,不得不谈AbstractQueuedSynchronizer(AQS)!
类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock/Semaphore/CountDownLatch...
并发包的底层就是使用AQS实现的,以下是AQS的类图结构
框架
它维护了一个volatile int state(代表共享资源)和一个FIFO线程等待队列(多线程竞争资源被阻塞会进入此队列)。这里volatile保证线程可见性。
state的访问方式有三种:
getState()
setState()
compareAndSetState()
这三种都是原子操作,其中compareAndSetState的实现依赖于Unsafe的compareAndSwapInt()方法。代码如下:
/** * 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. */ protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }
自定义资源共享方式
AQS定义了两种资源共享方式:Exclusive(独占,只有一个线程能执行,如ReentantLock)和Share(共享,多个线程可同时执行,如Semaphore/CountDownLatch)。
不同的自定义同步器争用共享资源的方式也不同,自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法。
isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int):独占方式。尝试获取资源,成功则返回true,失败则返回false。
tryRelease(int):独占方式。尝试释放资源,成功则返回true,失败则返回false。
tryAcquireShared(int):共享方式。尝试获取资源,负数表示失败;0表示成功,但没用剩余可用资源;正数表示成功,且有剩余资源。
tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待节点返回true,否则返回false。
源码实现
接下来我们开始开始讲解AQS的源码实现。依照acquire-release、acquireShared-releaseShared的次序来。
1. acquire(int)
acquire是一种以独占方式获取资源,如果获取到资源,线程直接返回,否则进入等待队列,直到获取到资源为止,且整个过程忽略中断的影响。该方法是独占模式下线程获取共享资源的顶层入口。
获取到资源后,线程就可以去执行其临界区代码了。下面是acquire()的源码
/** * Acquires in exclusive mode, ignoring interrupts. Implemented * by invoking at least once {@link #tryAcquire}, * returning on success. Otherwise the thread is queued, possibly * repeatedly blocking and unblocking, invoking {@link * #tryAcquire} until success. This method can be used * to implement method {@link Lock#lock}. * * @param arg the acquire argument. This value is conveyed to * {@link #tryAcquire} but is otherwise uninterpreted and * can represent anything you like. */ public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
通过注释我们知道,acquire方法是一种互斥模式,且忽略中断。该方法至少执行一次tryAcquire(int)
方法,如果tryAcquire(int)
方法返回true,则acquire直接返回,否则当前线程需要进入队列进行排队。函数流程如下
1、tryAcquire():尝试直接获取资源,如果成功则直接返回;
2、addWaiter():将该线程加入等待队列的尾部,并标记为独占模式;
3、acquireQueued():使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
4、如果线程在等待过程中被中断过,它是不响应的。只有获取资源后才再进行自我中断selfInterrupt(),将中断补上。
相关方法介绍
1.tryAcquire(int)
protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); }
tryAcquire尝试以独占的方式获取资源,如果获取成功,则直接返回true,否则直接返回false。该方法可以用于实现Lock中的tryLock()方法。该方法的默认实现是抛出UnsupportedOperationException异常,
具体实现由自定义的扩展了AQS的同步类来实现。AQS在这里只负责定义了一个公共的方法框架。这里之所以没用定义为abstract,是因为独占模式下只用实现tryAcquire-tryRelease,而共享模式下只用实现tryAcquireShared-tryReleaseShared
如果都定义成abstract,那么每个模式都要去实现另外一个模式下的接口。
2.addWaiter(Node)
/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */ 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.EXCLUSIVE
互斥模式、Node.SHARED
共享模式)加入到等待队列的队尾并返回当前线程所在的节点。如果队列不为空,
则以通过compareAndSetTail
方法以CAS的方式将当前节点加入到等待队列的末尾。否则,通过enq(node)方法初始化一个等待队列,并返回当前节点。
3. enq(node)
/** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node's predecessor */ 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; } } } }
enq(node)用于将当前节点插入到等待队列,如果队列为空,则初始化当前队列。整个过程以CAS自旋的方式进行,直到成功加入队尾为止。
4.acquireQueued(Node, int)
/** * Acquires in exclusive uninterruptible mode for thread already in * queue. Used by condition wait methods as well as acquire. * * @param node the node * @param arg the acquire argument * @return {@code true} if interrupted while waiting */ final boolean acquireQueued(final Node node, int arg) { boolean failed = true;//——标记是否成功拿到资源,默认是false 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()
用于队列中的线程自旋地以独占且不可中断的方式获取同步状态(acquire),直到拿到锁之后再返回。该方法的实现分成两部分:
如果当前节点已经成为头结点,尝试获取锁(tryAcquire)成功,然后返回;否则检查当前节点是否应该被park,然后将该线程park并且检查当前线程是否被可以被中断。
5.shouldParkAfterFailedAcquire(Node, 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 don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
shouldParkAfterFailedAcquire方法通过对当前节点的前一个节点的状态进行判断,对当前节点做出不同的操作,至于每个Node的状态表示,可以参考接口文档。
6.parkAndCheckInterrupt()
private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }
该方法让线程去休息,真正进入等待状态。park()会让当前线程进入waiting状态。在此状态下,有两种途径可以唤醒该线程:
1)被unpark();
2)被interrupt()。
需要注意的是,Thread.interrupted()会清除当前线程的中断标记位。
我们再回到acquireQueued(),总结下该函数的具体流程:
1、节点进入队尾后,检查状态,找到安全休息点
2、调用park()进入waiting状态,等待unpark()或interrupt()唤醒自己
3、被唤醒后,看自己是不是有资格能拿到号。如果能拿到,head指向当前节点,并返回从入队到拿到号的整个过程中是否被中断过;如果没用拿到,继续流程1
最后,总结一下acquire()的流程:
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
1、调用自定义同步器的tryAcquire()尝试直接去获取资源,如果成功则直接返回;
2、没成功,则执行addWaiter()将线程加入等待队列的尾部并标记为独占模式;
3、acquireQueued()使线程在等待队列中休息,有机会时(轮到自己,会被unpark())会去尝试获取资源。获取到资源才返回。如果在整个等待过程中被中断过,则会返回true,否则返回false。
4、如果线程在等待过程中被中断过,他是不响应的。只是获取资源后才进行自我中断selfInterrupt(),将中断补上。
未完...
参考书籍
Java并发编程之美
参考链接
https://www.jianshu.com/p/da9d051dcc3d