jdk的JUC包(java.util.concurrent)提供大量Java并发工具提供使用,基本由Doug Lea编写,很多地方值得学习和借鉴,是进阶升级必经之路
本文从JUC包中常用的对象锁、并发工具的使用和功能特性入手,带着问题,由浅到深,一步步剖析并发底层AQS抽象类具体实现
AQS是一个抽象类,类全路径java.util.concurrent.locks.AbstractQueuedSynchronizer,抽象队列同步器,是基于模板模式开发的并发工具抽象类,有如下并发类基于AQS实现:
CAS是Conmpare And Swap(比较和交换)的缩写,是一个原子操作指令
CAS机制当中使用了3个基本操作数:内存地址addr,预期旧的值oldVal,要修改的新值newVal 更新一个变量的时候,只有当变量的预期值oldVal和内存地址addr当中的实际值相同时,才会将内存地址addr对应的值修改为newVal
基于乐观锁的思路,通过CAS在不断尝试和比较,可以对变量值线程安全地更新
线程中断是一种线程协作机制,用于协作其他线程中断任务的执行
当线程处于阻塞等待状态,例如调用了wait()、join()、sleep()方法之后,调用线程的interrupt()方法之后,线程会马上退出阻塞并收到InterruptedException;
当线程处于运行状态,调用线程的interrupt()方法之后,线程并不会马上中断执行,需要在线程的具体任务执行逻辑中通过调用isInterrupted() 方法检测线程中断标志位,然后主动响应中断,通常是抛出InterruptedException
下面先介绍对象锁、并发工具有哪些基本特性,后面再逐步展开这些特性如何实现
以ReentrantLock锁为例,主要支持以下4种方式显式获取锁
ReentrantLock lock = new ReentrantLock(); // 一直阻塞等待,直到获取成功 lock.lock(); 复制代码
ReentrantLock lock = new ReentrantLock(); // 尝试获取锁,如果锁已被其他线程占用,则不阻塞等待直接返回false // 返回true - 锁是空闲的且被本线程获取,或者已经被本线程持有 // 返回false - 获取锁失败 boolean isGetLock = lock.tryLock(); 复制代码
ReentrantLock lock = new ReentrantLock(); try { // 尝试在指定时间内获取锁 // 返回true - 锁是空闲的且被本线程获取,或者已经被本线程持有 // 返回false - 指定时间内未获取到锁 lock.tryLock(10, TimeUnit.SECONDS); } catch (InterruptedException e) { // 内部调用isInterrupted() 方法检测线程中断标志位,主动响应中断 e.printStackTrace(); } 复制代码
ReentrantLock lock = new ReentrantLock(); try { // 响应中断获取锁 // 如果调用线程的thread.interrupt()方法设置线程中断,线程退出阻塞等待并抛出中断异常 lock.lockInterruptibly(); } catch (InterruptedException e) { e.printStackTrace(); } 复制代码
ReentrantLock lock = new ReentrantLock(); lock.lock(); // ... 各种业务操作 // 显式释放锁 lock.unlock(); 复制代码
已经获取到锁的线程,再次请求该锁可以直接获得
指同一个资源允许多个线程共享,例如读写锁的读锁允许多个线程共享,共享锁可以让多个线程并发安全地访问数据,提高程序执效率
公平锁:多个线程采用先到先得的公平方式竞争锁。每次加锁前都会检查等待队列里面有没有线程排队,没有才会尝试获取锁。 非公平锁:当一个线程采用非公平的方式获取锁时,该线程会首先去尝试获取锁而不是等待。如果没有获取成功,才会进入等待队列
因为非公平锁方式可以使后来的线程有一定几率直接获取锁,减少了线程挂起等待的几率,性能优于公平锁
类似Object的wait()、wait(long timeout)、notify()以及notifyAll()的方法结合synchronized内置锁可以实现可以实现等待/通知模式,实现Lock接口的ReentrantLock、ReentrantReadWriteLock等对象锁也有类似功能:
Condition接口定义了await()、awaitNanos(long)、signal()、signalAll()等方法,配合对象锁实例实现等待/通知功能,原理是基于AQS内部类ConditionObject实现Condition接口,线程await后阻塞并进入CLH队列(下面提到),等待其他线程调用signal方法后被唤醒
CLH队列,CLH是算法提出者Craig, Landin, Hagersten的名字简称
AQS内部维护着一个双向FIFO的CLH队列,AQS依赖它来管理等待中的线程,如果线程获取同步竞争资源失败时,会将线程阻塞,并加入到CLH同步队列;当竞争资源空闲时,基于CLH队列阻塞线程并分配资源
CLH的head节点保存当前占用资源的线程,或者是没有线程信息,其他节点保存排队线程信息
CLH中每一个节点的状态(waitStatus)取值如下:
AQS定义两种资源共享方式: Exclusive 独占,只有一个线程能执行,如ReentrantLock Share 共享,多个线程可同时执行,如Semaphore/CountDownLatch
AQS 基于sun.misc.Unsafe类提供的park方法阻塞线程,unpark方法唤醒线程,被park方法阻塞的线程能响应interrupt()中断请求退出阻塞
核心设计思路:AQS提供一个框架,用于实现依赖于CLH队列的阻塞锁和相关的并发同步器。 子类通过实现判定是否能获取/释放资源的protect方法,AQS基于这些protect方法实现对线程的排队、唤醒的线程调度策略
AQS还提供一个支持线程安全原子更新的int类型变量作为同步状态值(state),子类可以根据实际需求,灵活定义该变量代表的意义进行更新
通过子类重新定义的系列protect方法如下:
这些方法始终由需要需要调度协作的线程来调用,子类须以非阻塞的方式重新定义这些方法
AQS基于上述tryXXX方法,对外提供下列方法来获取/释放资源:
以独占模式为例:获取/释放资源的核心的实现如下:
Acquire: while (!tryAcquire(arg)) { 如果线程尚未排队,则将其加入队列; } Release: if (tryRelease(arg)) 唤醒CLH中第一个排队线程 复制代码
到这里,有点绕,下面一张图把上面介绍到的设计思路再重新捋一捋:
下面介绍基于AQS的对象锁、并发工具的一系列功能特性的实现原理
该特性还是以ReentrantLock锁为例,ReentrantLock是可重入对象锁,线程每次请求获取成功一次锁,同步状态值state加1,释放锁state减1,state为0代表没有任何线程持有锁
ReentrantLock锁支持公平/非公平特性,下面的显式获取特性以公平锁为例
基本实现如下:
ReentrantLock中的tryAcquire(int)方法实现:
protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); // 没有任何线程持有锁 if (c == 0) { // 通过CLH队列的head判断没有别的线程在比当前更早acquires // 且基于CAS设置state成功(期望的state旧值为0) if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { // 设置持有锁的线程为当前线程 setExclusiveOwnerThread(current); return true; } } // 持有锁的线程为当前线程 else if (current == getExclusiveOwnerThread()) { // 仅仅在当前线程,单线程,不用基于CAS更新 int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } // 其他线程已经持有锁 return false; } 复制代码
AQS的acquire(int)方法实现
public final void acquire(int arg) { // tryAcquire检查释放能获取成功 // addWaiter 构建CLH的节点对象并入队 // acquireQueued线程阻塞等待 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // acquireQueued返回true,代表线程在获取资源的过程中被中断 // 则调用该方法将线程中断标志位设置为true selfInterrupt(); } final boolean acquireQueued(final Node node, int arg) { // 标记是否成功拿到资源 boolean failed = true; try { // 标记等待过程中是否被中断过 boolean interrupted = false; // 循环直到资源释放 for (;;) { // 拿到前驱节点 final Node p = node.predecessor(); // 如果前驱是head,即本节点是第二个节点,才有资格去尝试获取资源 // 可能是head释放完资源唤醒本节点,也可能被interrupt() if (p == head && tryAcquire(arg)) { // 成功获取资源 setHead(node); // help GC p.next = null; failed = false; return interrupted; } // 需要排队阻塞等待 // 如果在过程中线程中断,不响应中断 // 且继续排队获取资源,设置interrupted变量为true if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } 复制代码
ReentrantLock中的tryLock()的实现仅仅是非公平锁实现,实现逻辑基本与tryAcquire一致,不同的是没有通过hasQueuedPredecessors()检查CLH队列的head是否有其他线程在等待,这样当资源释放时,有线程请求资源能插队优先获取
ReentrantLock中tryLock()具体实现如下:
public boolean tryLock() { return sync.nonfairTryAcquire(1); } final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); // 没有任何线程持有锁 if (c == 0) { // 基于CAS设置state成功(期望的state旧值为0) // 没有检查CLH队列中是否有线程在等待 if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } // 持有锁的线程为当前线程 else if (current == getExclusiveOwnerThread()) { // 仅仅在当前线程,单线程,不用基于CAS更新 int nextc = c + acquires; if (nextc < 0) // overflow,整数溢出 throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } // 其他线程已经持有锁 return false; } 复制代码
基本实现如下:
ReentrantLock中的实现如下:
public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireNanos(1, unit.toNanos(timeout)); } public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { // 如果线程已经被interrupt()方法设置中断 if (Thread.interrupted()) throw new InterruptedException(); // 先tryAcquire尝试获取锁 return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout); } 复制代码
AQS中的实现如下:
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(); // 如果前驱是head,即本节点是第二个节点,才有资格去尝试获取资源 // 可能是head释放完资源唤醒本节点,也可能被interrupt() if (p == head && tryAcquire(arg)) { // 成功获取资源 setHead(node); // help GC p.next = null; failed = false; return true; } // 更新剩余超时时间 nanosTimeout = deadline - System.nanoTime(); if (nanosTimeout <= 0L) return false; // 排队是否需要排队阻塞等待 // 且超时时间大于1微秒,则线程休眠到超时时间到了再尝试获取 if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); // 如果线程已经被interrupt()方法设置中断 // 则不再排队,直接退出 if (Thread.interrupted()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } 复制代码
ReentrantLock响应中断获取锁的方式是:当线程在park方法休眠中响应thead.interrupt()方法中断唤醒时,检查到线程中断标志位为true,主动抛出异常,核心实现在AQS的doAcquireInterruptibly(int)方法中
基本实现与阻塞等待获取类似,只是调用从AQS的acquire(int)方法,改为调用AQS的doAcquireInterruptibly(int)方法
private void doAcquireInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.EXCLUSIVE); // 标记是否成功拿到资源 boolean failed = true; try { for (;;) { // 拿到前驱节点 final Node p = node.predecessor(); // 如果前驱是head,即本节点是第二个节点,才有资格去尝试获取资源 // 可能是head释放完资源唤醒本节点,也可能被interrupt() if (p == head && tryAcquire(arg)) { // 成功获取资源 setHead(node); p.next = null; // help GC failed = false; return; } // 需要排队阻塞等待 if (shouldParkAfterFailedAcquire(p, node) && // 从排队阻塞中唤醒,如果检查到中断标志位为true parkAndCheckInterrupt()) // 主动响应中断 throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } 复制代码
AQS资源共享方式分为独占式和共享式,这里先以ReentrantLock为例介绍独占式资源的显式释放,共享式后面会介绍到
与显式获取有类似之处,ReentrantLock显式释放基本实现如下:
ReentrantLock中tryRelease(int)方法实现如下:
protected final boolean tryRelease(int releases) { int c = getState() - releases; // 只有持有锁的线程才有资格释放锁 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); // 标识是否没有任何线程持有锁 boolean free = false; // 没有任何线程持有锁 // 可重入锁每lock一次都需要对应一次unlock if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; } 复制代码
AQS中的release(int)方法实现如下:
public final boolean release(int arg) { // 尝试释放资源 if (tryRelease(arg)) { Node h = head; // 头节点不为空 // 后继节点入队后进入休眠状态之前,会将前驱节点的状态更新为SIGNAL(-1) // 头节点状态为0,代表没有后继的等待节点 if (h != null && h.waitStatus != 0) // 唤醒第二个节点 // 头节点是占用资源的线程,第二个节点才是首个等待资源的线程 unparkSuccessor(h); return true; } return false; } 复制代码
可重入的实现比较简单,以ReentrantLock为例,主要是在tryAcquire(int)方法中实现,持有锁的线程是不是当前线程,如果是,更新同步状态值state,并返回true,代表能获取锁
可共享资源以ReentrantReadWriteLock为例,跟独占锁ReentrantLock的区别主要在于,获取的时候,多个线程允许共享读锁,当写锁释放时,多个阻塞等待读锁的线程能同时获取到
ReentrantReadWriteLock类中将AQS的state同步状态值定义为,高16位为读锁持有数,低16位为写锁持有锁
ReentrantReadWriteLock中tryAcquireShared(int)、tryReleaseShared(int)实现的逻辑较长,主要涉及读写互斥、可重入判断、读锁对写锁的让步,篇幅所限,这里就不展开了
AQS中共享模式获取资源的具体实现如下:
public final void acquireShared(int arg) { // tryAcquireShared返回负数代表获取共享资源失败 // 则通过进入等待队列,直到获取到资源为止才返回 if (tryAcquireShared(arg) < 0) doAcquireShared(arg); } // 与前面介绍到的acquireQueued逻辑基本一致 // 不同的是将tryAcquire改为tryAcquireShared // 还有资源获取成功后将传播给CLH队列上等待该资源的节点 private void doAcquireShared(int arg) { 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); // 资源获取成功 if (r >= 0) { // 传播给CLH队列上等待该资源的节点 setHeadAndPropagate(node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; } } // 需要排队阻塞等待 // 如果在过程中线程中断,不响应中断 // 且继续排队获取资源,设置interrupted变量为true if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } // 资源传播给CLH队列上等待该资源的节点 private void setHeadAndPropagate(Node node, int propagate) { Node h = head; setHead(node); if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared()) // 释放共享资源 doReleaseShared(); } } 复制代码
释放读锁(ReadLock.unlock())主要实现如下: ReentrantReadWriteLock中共享资源的释放主要实现如下:
AQS中共享模式释放资源具体实现如下:
public final boolean releaseShared(int arg) { // 允许唤醒CLH中的休眠线程 if (tryReleaseShared(arg)) { // 执行资源释放 doReleaseShared(); return true; } return false; } private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; // 当前节点正在等待资源 if (ws == Node.SIGNAL) { // 当前节点被其他线程唤醒了 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; unparkSuccessor(h); } // 进入else的条件是,当前节点刚刚成为头节点 // 尾节点刚刚加入CLH队列,还没在休眠前将前驱节点状态改为SIGNAL // CAS失败是尾节点已经在休眠前将前驱节点状态改为SIGNAL else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; } // 每次唤醒后驱节点后,线程进入doAcquireShared方法,然后更新head // 如果h变量在本轮循环中没有被改变,说明head == tail,队列中节点全部被唤醒 if (h == head) break; } } 复制代码
这个特性实现比较简单,以ReentrantLock锁为例,公平锁直接基于AQS的acquire(int)获取资源,而非公平锁先尝试插队:基于CAS,期望state同步变量值为0(没有任何线程持有锁),更新为1,如果CAS更新失败在进行排队
// 公平锁实现 final void lock() { acquire(1); } // 非公平锁实现 final void lock() { // state值为0代表没有任何线程持有锁,直接插队获得锁 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } 复制代码
AQS的state变量值的含义不一定代表资源,不同的AQS的继承类可以对state变量值有不同的定义
例如在countDownLatch类中,state变量值代表还需释放的latch计数(可以理解为需要打开的门闩数),需要每个门闩都打开,门才能打开,所有等待线程才会开始执行,每次countDown()就会对state变量减1,如果state变量减为0,则唤醒CLH队列中的休眠线程
学习类似底层源码建议先定几个问题,带着问题学习;通俗学习前建议先理解透彻整体设计,整体原理(可以先阅读相关文档资料),再研究和源码细节,避免一开始就扎进去源码,容易无功而返