理解Java并发底层之AQS实现

jdk的JUC包(java.util.concurrent)提供大量Java并发工具提供使用，基本由Doug Lea编写，很多地方值得学习和借鉴，是进阶升级必经之路

本文从JUC包中常用的对象锁、并发工具的使用和功能特性入手，带着问题，由浅到深，一步步剖析并发底层AQS抽象类具体实现

名词解释

1 AQS

AQS是一个抽象类，类全路径java.util.concurrent.locks.AbstractQueuedSynchronizer，抽象队列同步器，是基于模板模式开发的并发工具抽象类，有如下并发类基于AQS实现：

2 CAS

CAS是Conmpare And Swap(比较和交换)的缩写，是一个原子操作指令

CAS机制当中使用了3个基本操作数：内存地址addr，预期旧的值oldVal，要修改的新值newVal 更新一个变量的时候，只有当变量的预期值oldVal和内存地址addr当中的实际值相同时，才会将内存地址addr对应的值修改为newVal

基于乐观锁的思路，通过CAS在不断尝试和比较，可以对变量值线程安全地更新

3 线程中断

线程中断是一种线程协作机制，用于协作其他线程中断任务的执行

当线程处于阻塞等待状态，例如调用了wait()、join()、sleep()方法之后，调用线程的interrupt()方法之后，线程会马上退出阻塞并收到InterruptedException；

当线程处于运行状态，调用线程的interrupt()方法之后，线程并不会马上中断执行，需要在线程的具体任务执行逻辑中通过调用isInterrupted() 方法检测线程中断标志位，然后主动响应中断，通常是抛出InterruptedException

对象锁特性

下面先介绍对象锁、并发工具有哪些基本特性，后面再逐步展开这些特性如何实现

1 显式获取

以ReentrantLock锁为例，主要支持以下4种方式显式获取锁

• (1) 阻塞等待获取

ReentrantLock lock = new ReentrantLock();
// 一直阻塞等待，直到获取成功
lock.lock();

• (2) 无阻塞尝试获取

ReentrantLock lock = new ReentrantLock();
// 尝试获取锁，如果锁已被其他线程占用，则不阻塞等待直接返回false
// 返回true - 锁是空闲的且被本线程获取，或者已经被本线程持有
// 返回false - 获取锁失败
boolean isGetLock = lock.tryLock();

• (3) 指定时间内阻塞等待获取

ReentrantLock lock = new ReentrantLock();
try {
    // 尝试在指定时间内获取锁
    // 返回true - 锁是空闲的且被本线程获取，或者已经被本线程持有
    // 返回false - 指定时间内未获取到锁
    lock.tryLock(10, TimeUnit.SECONDS);
} catch (InterruptedException e) {
    // 内部调用isInterrupted() 方法检测线程中断标志位，主动响应中断
    e.printStackTrace();
}

• (4) 响应中断获取

ReentrantLock lock = new ReentrantLock();
try {
    // 响应中断获取锁
    // 如果调用线程的thread.interrupt()方法设置线程中断，线程退出阻塞等待并抛出中断异常
    lock.lockInterruptibly();
} catch (InterruptedException e) {
    e.printStackTrace();
}

2 显式释放

ReentrantLock lock = new ReentrantLock();
lock.lock();
// ... 各种业务操作
// 显式释放锁
lock.unlock();

3 可重入

已经获取到锁的线程，再次请求该锁可以直接获得

4 可共享

指同一个资源允许多个线程共享，例如读写锁的读锁允许多个线程共享，共享锁可以让多个线程并发安全地访问数据，提高程序执效率

5 公平、非公平

公平锁：多个线程采用先到先得的公平方式竞争锁。每次加锁前都会检查等待队列里面有没有线程排队，没有才会尝试获取锁。非公平锁：当一个线程采用非公平的方式获取锁时，该线程会首先去尝试获取锁而不是等待。如果没有获取成功，才会进入等待队列

因为非公平锁方式可以使后来的线程有一定几率直接获取锁，减少了线程挂起等待的几率，性能优于公平锁

AQS实现原理

1 基本概念

(1) Condition接口

类似Object的wait()、wait(long timeout)、notify()以及notifyAll()的方法结合synchronized内置锁可以实现可以实现等待/通知模式，实现Lock接口的ReentrantLock、ReentrantReadWriteLock等对象锁也有类似功能：

Condition接口定义了await()、awaitNanos(long)、signal()、signalAll()等方法，配合对象锁实例实现等待/通知功能，原理是基于AQS内部类ConditionObject实现Condition接口，线程await后阻塞并进入CLH队列(下面提到)，等待其他线程调用signal方法后被唤醒

(2) CLH队列

CLH队列，CLH是算法提出者Craig, Landin, Hagersten的名字简称

AQS内部维护着一个双向FIFO的CLH队列，AQS依赖它来管理等待中的线程，如果线程获取同步竞争资源失败时，会将线程阻塞，并加入到CLH同步队列；当竞争资源空闲时，基于CLH队列阻塞线程并分配资源

CLH的head节点保存当前占用资源的线程，或者是没有线程信息，其他节点保存排队线程信息

CLH

CLH中每一个节点的状态(waitStatus)取值如下：

• CANCELLED(1)：表示当前节点已取消调度。当timeout或被中断（响应中断的情况下），会触发变更为此状态，进入该状态后的节点将不会再变化

• SIGNAL(-1)：表示后继节点在等待当前节点唤醒。后继节点入队后进入休眠状态之前，会将前驱节点的状态更新为SIGNAL

• CONDITION(-2)：表示节点等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的节点将从等待队列转移到同步队列中，等待获取同步锁

• PROPAGATE(-3)：共享模式下，前驱节点不仅会唤醒其后继节点，同时也可能会唤醒后继的后继节点

• 0：新节点入队时的默认状态

(3) 资源共享方式

AQS定义两种资源共享方式：Exclusive 独占，只有一个线程能执行，如ReentrantLock Share 共享，多个线程可同时执行，如Semaphore/CountDownLatch

(4) 阻塞/唤醒线程的方式

AQS 基于sun.misc.Unsafe类提供的park方法阻塞线程，unpark方法唤醒线程，被park方法阻塞的线程能响应interrupt()中断请求退出阻塞

2 基本设计

核心设计思路：AQS提供一个框架，用于实现依赖于CLH队列的阻塞锁和相关的并发同步器。子类通过实现判定是否能获取/释放资源的protect方法，AQS基于这些protect方法实现对线程的排队、唤醒的线程调度策略

AQS还提供一个支持线程安全原子更新的int类型变量作为同步状态值(state)，子类可以根据实际需求，灵活定义该变量代表的意义进行更新

通过子类重新定义的系列protect方法如下：

• boolean tryAcquire(int) 独占方式尝试获取资源，成功则返回true，失败则返回false

• boolean tryRelease(int) 独占方式尝试释放资源，成功则返回true，失败则返回false

• int tryAcquireShared(int) 共享方式尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源

• boolean tryReleaseShared(int) 共享方式尝试释放资源，如果释放后允许唤醒后续等待节点返回true，否则返回false

这些方法始终由需要需要调度协作的线程来调用，子类须以非阻塞的方式重新定义这些方法

AQS基于上述tryXXX方法，对外提供下列方法来获取/释放资源：

• void acquire(int) 独占方式获取到资源，线程直接返回，否则进入等待队列，直到获取到资源为止，且整个过程忽略中断的影响

• boolean release(int) 独占方式下线程释放资源，先释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源

• void acquireShared(int) 独占方式获取资源

• boolean releaseShared(int) 共享方式释放资源

以独占模式为例：获取/释放资源的核心的实现如下：

 Acquire:
     while (!tryAcquire(arg)) {
        如果线程尚未排队，则将其加入队列；
     }

 Release:
     if (tryRelease(arg))
        唤醒CLH中第一个排队线程

到这里，有点绕，下面一张图把上面介绍到的设计思路再重新捋一捋：

AQS基本设计

特性实现

下面介绍基于AQS的对象锁、并发工具的一系列功能特性的实现原理

1 显式获取

该特性还是以ReentrantLock锁为例，ReentrantLock是可重入对象锁，线程每次请求获取成功一次锁，同步状态值state加1，释放锁state减1，state为0代表没有任何线程持有锁

ReentrantLock锁支持公平/非公平特性，下面的显式获取特性以公平锁为例

(1) 阻塞等待获取

基本实现如下：

• 1、ReentrantLock实现AQS的tryAcquire(int)方法，先判断：如果没有任何线程持有锁，或者当前线程已经持有锁，则返回true，否则返回false

• 2、AQS的acquire(int)方法判断当前节点是否为head且基于tryAcquire(int)能否获得资源，如果不能获得，则加入CLH队列排队阻塞等待

• 3、ReentrantLock的lock()方法基于AQS的acquire(int)方法阻塞等待获取锁

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);
	}
}

(2) 无阻塞尝试获取

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;
}

(3) 指定时间内阻塞等待获取

基本实现如下：

• 1、ReentrantLock的tryLock(long, TimeUnit)调用AQS的tryAcquireNanos(int, long)方法

• 2、AQS的tryAcquireNanos先调用tryAcquire(int)尝试获取，获取不到再调用doAcquireNanos(int, long)方法

• 3、AQS的doAcquireNanos判断当前节点是否为head且基于tryAcquire(int)能否获得资源，如果不能获得且超时时间大于1微秒，则休眠一段时间后再尝试获取

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);
	}
}

(4) 响应中断获取

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);
	}
}

2 显式释放

AQS资源共享方式分为独占式和共享式，这里先以ReentrantLock为例介绍独占式资源的显式释放，共享式后面会介绍到

与显式获取有类似之处，ReentrantLock显式释放基本实现如下：

• 1、ReentrantLock实现AQS的tryRelease(int)方法，方法将state变量减1，如果state变成0代表没有任何线程持有锁，返回true，否则返回false

• 2、AQS的release(int)方法基于tryRelease(int)排队是否有任何线程持有资源，如果没有，则唤醒CLH队列中头节点的线程

• 3、被唤醒后的线程继续执行acquireQueued(Node,int)或者doAcquireNanos(int, long)或者doAcquireInterruptibly(int)中for(;;)中的逻辑，继续尝试获取资源

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;
}

3 可重入

可重入的实现比较简单，以ReentrantLock为例，主要是在tryAcquire(int)方法中实现，持有锁的线程是不是当前线程，如果是，更新同步状态值state，并返回true，代表能获取锁

4 可共享

可共享资源以ReentrantReadWriteLock为例，跟独占锁ReentrantLock的区别主要在于，获取的时候，多个线程允许共享读锁，当写锁释放时，多个阻塞等待读锁的线程能同时获取到

ReentrantReadWriteLock类中将AQS的state同步状态值定义为，高16位为读锁持有数，低16位为写锁持有锁

ReentrantReadWriteLock中tryAcquireShared(int)、tryReleaseShared(int)实现的逻辑较长，主要涉及读写互斥、可重入判断、读锁对写锁的让步，篇幅所限，这里就不展开了

获取读锁(ReadLock.lock())主要实现如下：

• 1、ReentrantReadWriteLock实现AQS的tryAcquireShared(int)方法，判断当前线程能否获得读锁

• 2、AQS的acquireShared(int)先基于tryAcquireShared(int)尝试获取资源，如果获取失败，则加入CLH队列排队阻塞等待

• 3、ReentrantReadWriteLock的ReadLock.lock()方法基于AQS的acquireShared(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*享资源的释放主要实现如下：

• 1、ReentrantReadWriteLock实现AQS的tryReleaseShared(int)方法，判断读锁释放后是否还有线程持有读锁

• 2、AQS的releaseShared(int)基于tryReleaseShared(int)判断是否需要CLH队列中的休眠线程，如果需要就执行doReleaseShared()

• 3、ReentrantReadWriteLock的ReadLock.unlock()方法基于AQS的releaseShared(int)方法释放锁

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;
	}
}

5 公平、非公平

这个特性实现比较简单，以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队列中的休眠线程

学习类似底层源码建议先定几个问题，带着问题来学习；通俗学习前建议先理解透彻整体设计，整体原理(可以先阅读相关文档资料)，再研究和源码细节，避免一开始就扎进去源码，容易无功而返