并发编程大师Doug Lea在编写JUC(java.util.concurrent)包的时候引入了java.util.concurrent.locks.AbstractQueuedSynchronizer,其实是Abstract Queued Synchronizer,也就是"基于队列实现的抽象同步器",一般我们称之为AQS。其实Doug Lea大神编写AQS是有严谨的理论基础的,他的个人博客上有一篇论文《The java.util.concurrent Synchronizer Framewor》,可以在互联网找到相应的译文《JUC同步器框架》,如果想要深入研究AQS必须要理解一下该论文的内容,然后结合论文内容详细分析一下AQS的源码实现。本文在阅读AQS源码的时候选用的JDK版本是JDK11。
出于写作习惯,下文会把AbstractQueuedSynchronizer称为AQS、JUC同步器框或者同步器框架。
AQS的主要功能AQS是JUC包中用于构建锁或者其他同步组件(信号量、事件等)的基础框架类。AQS从它的实现上看主要提供了下面的功能:
同步状态的原子性管理。
线程的阻塞和解除阻塞。
提供阻塞线程的存储队列。
基于这三大功能,衍生出下面的附加功能:
通过中断实现的任务取消,此功能基于线程中断实现。
可选的超时设置,也就是调用者可以选择放弃等待任务执行完毕直接返回。
定义了Condition接口,用于支持管程形式的await/signal/signalAll操作,代替了Object类基于JNI提供的wait/notify/notifyAll。
AQS还根据同步状态的不同管理方式区分为两种不同的实现:独占状态的同步器和共享状态的同步器。
同步器框架基本原理《The java.util.concurrent Synchronizer Framework》一文中其实有提及到同步器框架的伪代码:
// acquire操作如下: while (synchronization state does not allow acquire) { enqueue current thread if not already queued; possibly block current thread; } dequeue current thread if it was queued; //release操作如下: update synchronization state; if (state may permit a blocked thread to acquire){ unblock one or more queued threads; }撇脚翻译一下:
// acquire操作如下: while(同步状态申请获取失败){ if(当前线程未进入等待队列){ 当前线程放入等待队列; } 尝试阻塞当前线程; } 当前线程移出等待队列 //release操作如下: 更新同步状态 if(同步状态足够允许一个阻塞的线程申请获取){ 解除一个或者多个等待队列中的线程的阻塞状态; }为了实现上述操作,需要下面三个基本环节的相互协作:
同步状态的原子性管理。
等待队列的管理。
线程的阻塞与解除阻塞。
其实基本原理很简单,但是为了应对复杂的并发场景和并发场景下程序执行的正确性,同步器框架在上面的acquire操作和release操作中使用了大量的死循环和CAS等操作,再加上Doug Lea喜欢使用单行复杂的条件判断代码,如一个if条件语句会包含大量操作,AQS很多时候会让人感觉实现逻辑过于复杂。
同步状态管理AQS内部内部定义了一个32位整型的state变量用于保存同步状态:
/** * The synchronization state.(同步状态值) */ private volatile int state; // 获取state protected final int getState() { return state; } // 直接覆盖设置state protected final void setState(int newState) { state = newState; } // CAS设置state protected final boolean compareAndSetState(int expect, int update) { return STATE.compareAndSet(this, expect, update); }同步状态state在不同的实现中可以有不同的作用或者表示意义,这里其实不能单纯把它理解为中文意义上的"状态",它可以代表资源数、锁状态等等,下文遇到具体的场景我们再分析它表示的意义。
CLH队列与变体CLH锁即Craig, Landin, and Hagersten (CLH) locks,因为它底层是基于队列实现,一般也称为CLH队列锁。CLH锁也是一种基于链表的可扩展、高性能、公平的自旋锁,申请线程仅仅在本地变量上自旋,它不断轮询前驱的状态,假设发现前驱释放了锁就结束自旋。从实现上看,CLH锁是一种自旋锁,能确保无饥饿性,提供先来先服务的公平性。先看简单的CLH锁的一个简单实现:
public class CLHLock implements Lock { AtomicReference<QueueNode> tail = new AtomicReference<>(new QueueNode()); ThreadLocal<QueueNode> pred; ThreadLocal<QueueNode> current; public CLHLock() { current = ThreadLocal.withInitial(QueueNode::new); pred = ThreadLocal.withInitial(() -> null); } @Override public void lock() { QueueNode node = current.get(); node.locked = true; QueueNode pred = tail.getAndSet(node); this.pred.set(pred); while (pred.locked) { } } @Override public void unlock() { QueueNode node = current.get(); node.locked = false; current.set(this.pred.get()); } static class QueueNode { boolean locked; } // 忽略其他接口方法的实现 }