前言AbstractQueuedSynchronizer即队列同步器通过继承AbstractQueuedSynchronizer并重写其方法可以实现锁或其它同步组件本篇文章将对AbstractQueuedSynchronizer的使用和原理进行学习。参考资料《Java并发编程的艺术》正文一. AbstractQueuedSynchronizer的使用AbstractQueuedSynchronizer的使用通常如下。创建AbstractQueuedSynchronizer的子类作为同步组件例如ReentrantLock和CountDownLatch等的静态内部类并重写AbstractQueuedSynchronizer规定的可重写的方法同步组件通过调用AbstractQueuedSynchronizer提供的模板方法来实现同步组件的同步功能。先对AbstractQueuedSynchronizer的可重写的方法进行说明。AbstractQueuedSynchronizer是基于模板设计模式来实现锁或同步组件的AbstractQueuedSynchronizer内部维护着一个字段叫做state该字段表示同步状态是一个整型变量AbstractQueuedSynchronizer规定了若干方法来操作state字段但AbstractQueuedSynchronizer本身并没有对这些方法进行实现而是要求AbstractQueuedSynchronizer的子类来实现这些方法下面看一下这些方法的签名和注释。方法签名注释protected boolean tryAcquire(int arg)独占式地获取同步状态。该方法在独占式获取同步状态以前应该判断是否允许独占式获取如果允许则尝试基于CAS方式来设置同步状态设置成功则表示获取同步状态成功。protected boolean tryRelease(int arg)独占式地释放同步状态。可以理解为将同步状态还原为获取前的状态。protected int tryAcquireShared(int arg)共享式地获取同步状态。protected boolean tryReleaseShared(int arg)共享式地释放同步状态。protected boolean isHeldExclusively()判断当前线程是否独占当前队列同步器。实际上AbstractQueuedSynchronizer规定的这些可重写的方法均会被AbstractQueuedSynchronizer提供的模板方法所调用在基于AbstractQueuedSynchronizer实现同步组件时可根据同步组件的实际功能来重写这些可重写方法然后再通过调用模板方法来实现同步组件的功能。下面看一下AbstractQueuedSynchronizer提供的模板方法。方法签名注释public final void acquire(int arg)独占式获取同步状态即独占式获取锁。获取成功则该方法返回获取失败则当前线程进入同步队列等待。public final void acquireInterruptibly(int arg) throws InterruptedException独占式获取同步状态并响应中断。即如果获取同步状态失败则会进入同步队列等待此时如果线程被中断则会退出等待状态并抛出中断异常。public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException同acquireInterruptibly(int arg)并在其基础上指定了等待时间若超时还未获取同步状态则返回false。public final void acquireShared(int arg)共享式获取同步状态即共享式获取锁。获取成功则该方法返回获取失败则当前线程进入同步队列等待支持同一时刻多个线程获取到同步状态。public final void acquireSharedInterruptibly(int arg) throws InterruptedException共享式获取同步状态并响应中断。public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException同acquireSharedInterruptibly(int arg)并在其基础上指定了等待时间若超时还未获取同步状态则返回false。public final boolean release(int arg)独占式地释放同步状态即独占式地释放锁。成功释放锁之后会将同步队列中的第一个节点的线程唤醒。public final boolean releaseShared(int arg)共享式地释放同步状态。public final CollectionThread getQueuedThreads()获取在同步队列上等待地线程。下面结合《Java并发编程的艺术》中的例子的简化版来直观的展示使用AbstractQueuedSynchronizer来实现同步组件的方便如下所示。public class MyLock { private final Sync sync new Sync(); public void lock() { sync.acquire(1); } public void unlock() { sync.release(1); } private static class Sync extends AbstractQueuedSynchronizer { Override protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } Override protected boolean tryRelease(int unused) { if (getState() 0) { throw new IllegalMonitorStateException(); } setExclusiveOwnerThread(null); setState(0); return true; } } }上述的MyLock是一个提供了最简单功能的不可重入锁该锁的Lock()和unlock()方法全部是调用的AbstractQueuedSynchronizer提供的模板方法acquire()和release()然后在acquire()和release()这两个模板方法中又会调用Sync重写的tryAcquire()和tryRelease()方法最终的效果就是仅仅重写了tryAcquire()和tryRelease()方法便实现了一个具有最简单功能的不可重入锁而同步的具体实现细节比如同步队列等全都由AbstractQueuedSynchronizer来完成借助AbstractQueuedSynchronizer可以极大降低实现一个同步组件的门槛。二. AbstractQueuedSynchronizer的原理在AbstractQueuedSynchronizer中维护了一个FIFO的同步队列当某个线程获取同步状态失败时该线程会被封装成一个节点Node并加入同步队列中。本小节将结合同步队列和AbstractQueuedSynchronizer源码对AbstractQueuedSynchronizer的原理进行学习。在AbstractQueuedSynchronizer中有两个字段head和tail分别指向同步队列头节点和尾节点同步队列中的每个节点也有两个字段prev和next分别指向上一节点和下一节。一个同步队列的基本结构如下所示。当有新节点入队列时新节点会加入到队列尾并成为新的尾节点同时tail字段会指向新的尾节点。节点入队列如下所示。通常head字段指向的头节点表示当前成功获取同步状态的线程所对应的节点头节点释放同步状态后会唤醒头节点的下一节点当下一节点成功获取同步状态后会将head字段指向自己从而成为新的头节点与此同时新头节点的prev和老头节点的next字段会被置为null以帮助垃圾回收。头节点释放同步状态如下所示。上述为同步队列的基本概览下面将结合AbstractQueuedSynchronizer所提供的模板方法来对独占式获取释放同步状态和共享式获取释放同步状态进行分析。1. 独占式获取同步状态独占式获取同步状态的acquire()方法如下所示。public final void acquire(int arg) { if (!tryAcquire(arg) acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }acquire()方法做了如下几件事情先调用重写的tryAcquire()方法独占式地获取同步状态获取成功则返回获取失败则执行下一步骤调用addWaiter()方法基于当前线程创建一个节点Node并添加到同步队列中调用acquireQueued()方法使刚创建并加入了同步队列的Node进入自旋状态在自旋状态中Node会判断自己的上一节点是否是头节点如果是则尝试获取同步状态获取成功则退出自旋状态如果自己的上一节点不是头节点或者获取同步状态失败则阻塞自己直到被上一节点唤醒或者被中断此时重复前面的自旋过程。还是从源码入手分析创建节点Node入同步队列和自旋判断并获取同步状态这几个过程。首先是addWaiter()方法如下所示。private Node addWaiter(Node mode) { // 基于当前线程创建一个节点 Node node new Node(Thread.currentThread(), mode); Node pred tail; if (pred ! null) { node.prev pred; // 先尝试安全地将新创建的节点添加到同步队列尾 // 如果添加成功则直接返回新创建的节点 if (compareAndSetTail(pred, node)) { pred.next node; return node; } } // 如果新创建的节点是入同步队列的第一个节点或者尝试添加失败 // 则调用enq()方法将节点安全地添加到同步队列中 enq(node); return node; }在addWaiter()中基于当前线程创建节点Node之后会立即尝试一次安全地将新创建的Node添加到队列尾若失败的话则会调用enq()方法来让节点入队列enq()的实现如下所示。private Node enq(final Node node) { for (;;) { Node t tail; if (t null) { // node如果是入同步队列的第一个节点此时t为null // 则先将head安全的指向一个空节点然后head再赋值给tail // 此时头节点和尾节点均为同一个空节点 if (compareAndSetHead(new Node())) tail head; } else { // 尾节点不为null时则尝试安全的将node设置为新的尾节点 // 如果设置失败则循环重复尝试设置直到设置成功为止 node.prev t; if (compareAndSetTail(t, node)) { t.next node; return t; } } } }enq()方法实际就是在一个死循环中重复地基于CAS方式将新创建的节点入同步队列。成功入队列后的节点会被传入acquireQueued()方法进入自旋状态其实现如下所示。final boolean acquireQueued(final Node node, int arg) { boolean failed true; try { boolean interrupted false; for (;;) { // p为当前节点的上一节点 final Node p node.predecessor(); // 判断当前节点的上一节点是否是头节点 // 如果是则尝试获取同步状态 if (p head tryAcquire(arg)) { // 获取同步状态成功则将当前节点置为头节点 setHead(node); p.next null; failed false; return interrupted; } // 先判断当前节点是否应该被阻塞 // 如果应该被阻塞则调用parkAndCheckInterrupt()方法进入阻塞状态 // 否则自旋重新进行上述步骤 if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) interrupted true; } } finally { if (failed) cancelAcquire(node); } }acquireQueued()方法中如果当前节点的上一节点不是头节点或者尝试获取同步状态失败那么会在shouldParkAfterFailedAcquire()方法中判断当前节点的上一节点状态是否为SIGNAL如果是则会调用parkAndCheckInterrupt()进入阻塞状态如果不是则会将当前节点的上一节点状态置为SIGNAL然后自旋地再执行上述步骤。独占式获取同步状态还有两个模板方法分别为acquireInterruptibly()和tryAcquireNanos()其中acquireInterruptibly()可以响应中断tryAcquireNanos()可以指定等待时间下面将对这两个方法进行简要分析。acquireInterruptibly()的实现如下所示。public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (!tryAcquire(arg)) doAcquireInterruptibly(arg); }acquireInterruptibly()方法中尝试获取同步状态失败时会调用doAcquireInterruptibly()方法来将当前线程加入同步队列如下所示。private void doAcquireInterruptibly(int arg) throws InterruptedException { // 基于当前线程创建节点并加入同步队列尾 final Node node addWaiter(Node.EXCLUSIVE); boolean failed true; try { for (;;) { final Node p node.predecessor(); if (p head tryAcquire(arg)) { setHead(node); p.next null; failed false; return; } if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) // 等待过程中被中断则抛出中断异常 throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }doAcquireInterruptibly()的实现基本与acquireQueued()相同区别在于在doAcquireInterruptibly()方法中进入等待状态时被中断的话会抛出中断异常。接下来再看一下tryAcquireNanos()方法的实现如下所示。public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout); }tryAcquireNanos()方法中获取同步状态失败时会调用doAcquireNanos()方法来将当前线程加入同步队列如下所示。private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { if (nanosTimeout 0L) return false; // 计算等待的最晚时间点 final long deadline System.nanoTime() nanosTimeout; // 基于当前线程创建节点并加入同步队列尾 final Node node addWaiter(Node.EXCLUSIVE); boolean failed true; try { for (;;) { final Node p node.predecessor(); if (p head tryAcquire(arg)) { setHead(node); p.next null; failed false; return true; } // 计算剩余等待时间 nanosTimeout deadline - System.nanoTime(); if (nanosTimeout 0L) return false; // 如果当前线程应该被阻塞并且剩余等待时间大于1000纳秒则使用LockSupport进入等待状态 if (shouldParkAfterFailedAcquire(p, node) nanosTimeout spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted()) // 如果当前线程是因为被中断而从等待状态返回则抛出中断异常 throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }doAcquireNanos()方法的大体实现与acquireQueued()相同不同之处在于首先doAcquireNanos()至多只会在同步队列中阻塞nanosTimeout的时间超时的话会返回false表示等待获取同步状态失败其次doAcquireNanos()也是响应中断的。2. 独占式释放同步状态AbstractQueuedSynchronizer提供了模板方法release()来实现独占式释放同步状态如下所示。public final boolean release(int arg) { // 尝试释放同步状态 if (tryRelease(arg)) { Node h head; if (h ! null h.waitStatus ! 0) // 释放同步状态之后唤醒头节点的下一节点 unparkSuccessor(h); return true; } return false; }当前节点独占式地释放同步状态之后会调用unparkSuccessor()方法来唤醒下一节点如下所示。private void unparkSuccessor(Node node) { int ws node.waitStatus; if (ws 0) compareAndSetWaitStatus(node, ws, 0); Node s node.next; // 如果下一节点为null或者下一节点状态为关闭则从尾节点开始向前寻找满足唤醒条件的节点并唤醒 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); }unparkSuccessor()方法会获取当前节点的下一节点并判断下一节点是否为null或者状态是否为CANCELLED如果符合其中任何一项表明当前节点的下一节点不满足唤醒条件那么需要从尾节点开始向前寻找满足唤醒条件的节点并唤醒而唤醒条件就是节点不为null以及节点状态不为CANCELLED。3. 共享式获取同步状态共享式获取同步状态的模板方法acquireShared()如下所示。public final void acquireShared(int arg) { // 获取同步状态失败则返回负值 if (tryAcquireShared(arg) 0) // 创建节点加入同步队列尾并进入自旋状态 doAcquireShared(arg); }tryAcquireShared()方法对其返回值进行了如下规定。返回负值表示共享式获取同步状态失败返回正值表示共享式获取同步状态成功并且后续共享式获取同步状态还能成功返回0表示共享式获取同步状态成功但后续共享式获取同步状态会失败。因此如果通过tryAcquireShared()方法共享式获取同步状态失败则会调用doAcquireShared()方法基于当前线程创建节点添加到同步队列尾并进入自旋状态tryAcquireShared()方法如下所示。private void doAcquireShared(int arg) { // 基于当前线程和Node.SHARED创建一个节点Node并加入同步队列 // Node.SHARED实际上是一个空Node对象用于指示当前创建的节点是在共享模式下进行等待 final Node node addWaiter(Node.SHARED); boolean failed true; try { boolean interrupted false; for (;;) { final Node p node.predecessor(); // 获取当前节点的上一节点并判断是否是头节点 if (p head) { // 当前节点的上一节点是头节点则尝试共享式地获取同步状态 int r tryAcquireShared(arg); // tryAcquireShared()方法返回非负值表示获取同步状态成功 if (r 0) { // 将当前节点设置为头节点并判断下一节点是否是在共享模式下进行等待如果是则唤醒 setHeadAndPropagate(node, r); p.next null; if (interrupted) selfInterrupt(); failed false; return; } } if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt()) interrupted true; } } finally { if (failed) cancelAcquire(node); } }在doAcquireShared()方法中首先会基于当前线程创建一个节点Node并会设置节点的nextWaiter字段为Node.SHARED其中nextWaiter是用于指示当前节点是独占模式下等待的节点还是共享模式下等待的节点。创建好节点并添加到同步队列尾之后节点就会进入自旋状态整个自旋流程与独占式获取同步状态大体一致而区别就在于自旋状态中节点成功获取到了同步状态之后的操作这里如果在自旋状态中成功获取到了同步状态那么就会对获取到同步状态的节点的下一节点进行判断如果判断得到下一节点是在共享模式下等待的节点则需要将其唤醒整个判断是在setHeadAndPropagate()方法中实现的如下所示。private void setHeadAndPropagate(Node node, int propagate) { Node h head; setHead(node); // propagate为tryAcquireShared()方法的返回值 // propagate为负值表示共享式获取同步状态失败 // propagate为0表示共享式获取同步状态成功但是同步状态后续无法再被获取 // propagate为正值表示共享式获取同步状态失败同步状态后续还可以被获取 if (propagate 0 || h null || h.waitStatus 0 || (h head) null || h.waitStatus 0) { Node s node.next; // 如果当前节点的下一节点是共享模式下等待的节点则将其唤醒 if (s null || s.isShared()) doReleaseShared(); } }setHeadAndPropagate()方法中会先将共享式获取到同步状态的节点置为头节点然后如果同步状态后续还可以被获取并且节点的下一节点是在共享模式下等待的节点那么调用doReleaseShared()方法来唤醒共享模式下等待的节点如下所示。private void doReleaseShared() { for (;;) { // h表示本次循环执行时的头节点 Node h head; if (h ! null h ! tail) { int ws h.waitStatus; if (ws Node.SIGNAL) { // 如果头节点状态为SIGNAL表示头节点的下一节点处于等待状态 // 则先将头节点状态置为0再将下一节点唤醒 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; unparkSuccessor(h); } else if (ws 0 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; } if (h head) break; } }doReleaseShared()方法会每次获取循环执行时的头节点因为节点状态为SIGNAL时节点的下一节点处于等待状态所以如果头节点状态为SIGNAL则会先置头节点状态为0然后唤醒下一节点下一节点被唤醒后进行自旋时如果共享式地获取到了同步状态那么又会执行一轮上述的步骤也就是说共享模式下等待的节点的唤醒会发生在同步状态释放后和上一节点获取到同步状态后与之相比的独占模式下等待的节点的唤醒只会发生在同步状态释放后。上面的整个流程以一个简单的例子加以说明下图为一个同步队列的节点示意图。Node1在自旋过程中判断上一节点是头节点并且获取同步状态成功此时Node1成为头节点并判断同步状态还能继续获取以及下一节点Node2是在共享模式下等待的节点所以Node2被唤醒Node2被唤醒后进入自旋状态并判断上一节点Node1是头节点然后获取同步状态并成功此时Node2成为头节点因此在同步状态可以继续获取以及下一节点是共享模式下等待的节点的情况下共享模式下等待的节点均会被唤醒直到同步状态无法再被获取或者下一节点是独占模式下等待的节点。共享式获取同步状态还有两个方法acquireSharedInterruptibly()和tryAcquireSharedNanos()相较于acquireShared()方法分别增加了响应中断和指定等待时间的功能除此之外这两个方法的实现与acquireShared()一致故这里不再讨论这两个方法了。4. 共享式释放同步状态共享式释放同步状态的模板方法releaseShared()如下所示。public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }即调用重写的tryReleaseShared()方法共享式地释放同步状态成功之后会调用doReleaseShared()方法而前面已经知道doReleaseShared()方法会每次获取循环执行时的头节点然后唤醒头节点的下一节点如果下一节点是共享模式下等待的节点那么这个唤醒动作会被传递下去如果下一节点是独占模式下等待的节点则唤醒的动作只会作用于这个独占模式下等待的节点。总结AbstractQueuedSynchronizer提供了用于构建锁或者同步组件的基础框架通常使用方式为创建一个AbstractQueuedSynchronizer的子类并作为同步组件的静态内部类然后重写其规定的可重写方法然后同步组件功能的实现需要通过调用AbstractQueuedSynchronizer提供的模板方法这些模板方法会调用其规定的可重写方法最终可以按照我们的预期来实现同步功能。AbstractQueuedSynchronizer提供的模板方法主要分为独占式获取同步状态独占式释放同步状态共享式获取同步状态和共享式释放同步状态这些模板方法为我们提供了具体的同步实现细节比如等待和唤醒我们仅需将其规定的可重写方法重写好便能可靠地实现一个自定义同步组件。