AbstractQueuedSynchronizer
从锁说起
juc.locks包下提供了常见的锁相关的工具，用来替代synchronized关键字（jdk1.5之前synchronized效率较低）和加强一些锁功能，比如重入锁，公平非公平锁，读写锁等等。
AbstractQueuedSynchronizer简称aqs是juc.locks包中锁相关的一个最核心的类，ReentrantLock等锁相关的实现都离不开它。字面意思"抽象的排队的同步器"。
先抛开aqs，先尝试自己实现一个锁。
public class Lock {
private AtomicBoolean state = new AtomicBoolean();
public void lock() {
while (!state.compareAndSet(false, true)) {}
}
public void unLock() {
state.set(false);
}
}
如上，我们实现了一个最简单的锁，但是这个锁有个很大的缺点，线程A加完锁之后，线程B虽然加锁的时候会被阻塞（while循环），但是线程B却可以解锁，即使锁不是由B加上的。
我们预期的应该是锁由A加上，只能由A解锁。为此，需要稍微改动一下代码。
public class Lock2 {
private AtomicBoolean state = new AtomicBoolean();
private Thread ownThread;
public void lock() {
while (!state.compareAndSet(false, true)) {}
ownThread = Thread.currentThread();
}
public void unLock() {
if (ownThread != Thread.currentThread()) {
state.set(false);
ownThread = null;
}
}
}
我们添加了一个变量ownThread用来保存加锁的线程。
还有一个问题，锁是不可重入的，即一个线程只能加锁一次，同一个线程再次加锁同样会阻塞，导致程序无法运行。这显然不是我们想要的。
再优化一下。
public class Lock3 {
private volatile int state;
private AtomicReference<Thread> ownThread = new AtomicReference<>();
public void lock() {
Thread currentThread = Thread.currentThread();
if (ownThread.get() != currentThread) {
while (!ownThread.compareAndSet(null, currentThread)) {}
}
state++;
}
public void unLock() {
if (ownThread.get() == Thread.currentThread()) {
state--;
if (state == 0) {
ownThread.set(null);
}
} else {
throw new IllegalMonitorStateException();
}
}
}
我们用state存储了线程重入的次数，ownThread用来存储加锁的线程。这样貌似比较完美了，但是锁效率比较低，我们的锁是通过自旋来实现阻塞的。有没有更好的办法？更为高效的做法是线程获取不到锁，进入wait状态。线程解锁的时候，通知其他线程退出wait状态，并使被唤醒的线程重新尝试获取锁。我们再优化一下代码。
public class Lock4 {
private volatile int state;
private AtomicReference<Thread> ownThread = new AtomicReference<>();
private Set<Thread> threadHolder = Collections.synchronizedSet(new HashSet<>());
public void lock() {
Thread currentThread = Thread.currentThread();
if (ownThread.get() != currentThread) {
while (!ownThread.compareAndSet(null, currentThread)) {
threadHolder.add(currentThread);
LockSupport.park();
}
}
state++;
}
public void unLock() {
Thread currentThread = Thread.currentThread();
if (ownThread.get() == currentThread) {
state--;
if (state == 0) {
ownThread.set(null);
threadHolder.remove(currentThread);
threadHolder.stream().findAny().ifPresent(LockSupport::unpark);
}
} else {
throw new IllegalMonitorStateException();
}
}
}
这里用一个Set来保存所有尝试加锁但是失败并且阻塞的线程，这里的Set是线程安全的，我们不讨论线程安全的Set是怎么实现的以及效率问题，这里只是为了说明问题。
加锁失败的时候我们将当前线程加入集合，并阻塞。释放锁的时候，随机获取集合中的一个线程，唤醒。
还有没有优化的空间或者缺失的功能呢？
-
我们的锁不是公平锁
这个解决起来比较简单，将Set改成队列，维持等待线程的顺序，唤醒的时候唤醒队列头的线程。
-
我们的锁不能快失败，即获取不到锁时立刻返回false
这个也比较简单，获取不到锁的时候当前线程不加入队列即可。
-
我们的锁不支持共享锁
实现这个需要我们加入队列的不只是当前线程，还要维护一些额外的状态，比如能否共享等等。
来总结一下要实现所有功能，我们的锁需要有哪些必要条件。
- 一个state变量，用来存储锁的状态
- 一个Thread变量，用来存储持有锁的线程信息
-
一个队列，用来存放没有获得锁的线程信息
而AbstractQueuedSynchronizer帮我们抽象出了这些条件，提供了一些基本的方法帮助我们操作这些变量，我们可以很容易的通过这些方法实现自己的锁。
AbstractQueuedSynchronizer源码解析
可以看到AbstractQueuedSynchronizer继承于AbstractOwnableSynchronizer，AbstractQueuedLongSynchronizer和AbstractQueuedSynchronizer基本一样，只是state变量一个是int型一个是long型，我们以AbstractQueuedLongSynchronizer为例来分析源码。
AbstractOwnableSynchronizer
public abstract class AbstractOwnableSynchronizer
implements java.io.Serializable {
private static final long serialVersionUID = 3737899427754241961L;
protected AbstractOwnableSynchronizer() { }
private transient Thread exclusiveOwnerThread;
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
protected final Thread getExclusiveOwnerThread() {
return exclusiveOwnerThread;
}
}
类很简单，只有exclusiveOwnerThread一个变量，以及相应的get,set方法，这个变量就对应上边实现的锁的ownThread变量，用来存储得到锁的线程。
AbstractQueuedSynchronizer
变量
先来看看AbstractQueuedSynchronizer中的变量。
以Offsett结尾的变量是相应变量对应的偏移量，用来做cas操作，不用关注。Unsafe是jdk提供的操作底层的类,这里主要用来做cas操作，也不用关注。
我们最需要关注的是 head, tail,state变量。
head和tail组成了一个队列，对应我们自己实现的锁的队列。
state对应我们自己实现的锁的state。
可见aqs和我们实现的锁很类似，的确是将实现锁的公共条件抽象出来，来简化锁的实现。
head和tail的类型Node是个内部类，先来看看这是个什么。
static final class Node {
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
final boolean isShared() {
return nextWaiter == SHARED;
}
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node() { // Used to establish initial head or SHARED marker
}
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
prev和next变量不必多说，用来组成双向链表进而实现队列。thread用来存储等待中的线程的信息，waitStatus用来存储当前等待线程的一些额外的一些状态，这里可能取值1,0，-1，-2，-3，分别对应CANCELLED，默认，SIGNAL，CONDITION ，PROPAGATE。先不用关注这里的取值的具体的含义，之后我们再分析，只需要了解这里存储了线程的额外的一些状态。SHARED，EXCLUSIVE ，nextWaiter这些都是Node类型的变量，暂时也不做分析。
方法
接下来看看AQS提供的方法，AQS方法有很多，先来看几个基本方法，这些方法是aqs实现的基础。
/**
* 通过cas操作来设置头结点
*/
private final boolean compareAndSetHead(Node update) {
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
/**
* 通过cas操来设置尾结点
*/
private final boolean compareAndSetTail(Node expect, Node update) {
return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
/**
* 通过cas操作来设置状态，这里方法是protected，说明是为了提供给子类调用或者覆写
*/
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
/**
* 通过case操作某个结点的next变量的值
*/
private static final boolean compareAndSetNext(Node node,
Node expect,
Node update) {
return unsafe.compareAndSwapObject(node, nextOffset, expect, update);
}
/**
* 通过case操作某个结点的waitStatus
*/
private static final boolean compareAndSetWaitStatus(Node node,
int expect,
int update) {
return unsafe.compareAndSwapInt(node, waitStatusOffset,
expect, update);
}
/**
* 设置头部结点
*/
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
/**
* 设置状态
*/
protected final void setState(int newState) {
state = newState;
}
/**
* 利用上面提供的方法，实现线程安全的入队操作
*/
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;
}
}
}
}
/**
* 入队一个新结点到队尾，并将新结点的nextWaiter赋值为mode
*/
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
// 这里先尝试进行cas操作，如果失败再进入到复杂的enq(node)方法内，这种做法是为了效率考虑，在aqs的设计中，处处可以看到这些优化。
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
接下来回顾一下上面自己实现的锁，并抽象一下加锁的步骤。
- 尝试加锁并将state+1
- 加锁成功完成并退出，加锁失败到3
- 将当前线程加入队列，并阻塞当前尝试加锁的线程
解锁的步骤。
- 比对当前线程是否是加锁的线程
- 如果不是抛异常，如果是到3
- 将state-1，如果state为0到4，否则完成
- 从队列中拿出头结点对应的线程并唤醒
aqs将加锁抽象成一个方法 acquire()
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
其中tryAcquire()并没有具体的实现，交由子类来实现。可以在tryAcquire进行更改state的操作，一般需要在tryAcquire判断得到锁的线程是不是当前线程以及更新state
acquireQueued() 不需要子类来实现，aqs已经实现，主要作用在队列中添加结点，并阻塞。
这两步这和我们上边分析出的加锁的步骤高度一致。
当没有获取到锁，并且添加等待队列失败,假如期间线程被中断过，会执行selfInterrupt()，重新中断线程。这里很容易让人误解为acquire()是响应中断的（假如加锁的过程中，线程中断了，立刻终止加锁操作，并抛出异常），但其实并不是。
我们来看一下selfInterrupt()方法。
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
只是将当前的线程的中断标致重新置为已中断。
接下来带着两个疑问来分析acquireQueued()方法，是如何阻塞线程的以及为什么不响应中断。
需要明确的是，执行到此方法时，代表尝试获取锁失败(tryAcquire() == true)并且当前线程对应的Node已经加入到队尾(addWaiter())。
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获得当前线程对应结点的前驱结点
final Node p = node.predecessor();
//step1
//当前线程对应的结点是第二个结点，表明之前结点对应的线程的已经进行过解锁操作,此时尝试加锁,假如成功后，将当前结点置为头结点
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//step2
//假如当前线程对应的结点不是第二个结点或者没有抢到锁（举例线程A为头结点获得锁，线程B为第二个结点，当线程A释放锁的时候，线程C插队获得锁，此时线程B就没有获取到锁）,
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
//取消
cancelAcquire(node);
}
}
接下来依次分析shouldParkAfterFailedAcquire(),parkAndCheckInterrupt()方法。
/**
* 通过方法名称很容易判断出这个方法的作用是为了判断当前结点对应的线程是不是应该进入等待状态。
* 思考一下，什么情况下线程不能该进入wait状态？
* 两个线程A和B, A获得锁，B线程请求锁失败，尝试wait时，A线程进行解锁操作，并唤醒B线程，B线程此时wait，将永远无法被唤醒。
*
* 但是此方法并没有保证上述情况不发生，即（线程在没有进入wait转态的情况下被唤醒，之后又进入wait状态），因为shouldParkAfterFailedAcquire和parkAndCheckInterrupt并不是原子操作
* 但事实并没有这种情况，这个之后我们再分析。
* 这里我们假定shouldParkAfterFailedAcquire和parkAndCheckInterrupt是原子操作。
*
* 对于这个方法，当前能进入wait状态条件是，当前结点对应的第一个不是取消状态的结点[pre]的状态是SIGNAL，即-1
* 否则的话，不能进入wait状态。
*
*/
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.
* 前一个结点的waitStatus是SIGNAL的话，即-1，代表可以阻塞当前线程。
*
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
* 当前结点的前一个结点是取消状态 waitStatus > 0，此时把中间取消的结点“断掉”。
*/
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.
* 到这里的状态只能是0或者PROPAGATE，表明可以进入wait状态。
* 为什么只能是0或者PROPAGATE，但是却没有CONDITION状态呢，原因是CONDITION状态根本不会在这个队列里出现
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
/**
* 这个方法的作用就是使当前线程进入wait状态，从这里可以看到并没有响应中断，只是在阻塞被唤醒后返回了线程的中断状态。
*/
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
接下来我们来看relese方法，即抽象出来的解锁方法
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
同样 tryRelease()是个抽象方法，一般在此方法中进行更改状态操作，正常情况下都会返回true,因为解锁操作只有当前活动的线程在调用，不存在竞态。
unparkSuccessor()用来唤醒队列中下一个结点。
/**
* 唤醒线程
*/
private void unparkSuccessor(Node node) {
/*
* 将头结点状态更改为0,结合shouldParkAfterFailedAcquire()来看
* 线程要进入wait的条件，是前一个结点状态为SIGNAL
* 当头结点的状态为0时，头结点之后的结点必定不能进入wait状态
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*
* 找到头结点之后第一个不是取消状态的结点
*/
Node s = node.next;
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);
}
为了更好的讲解，我们模拟两个线程获取锁时，程序的执行流程，获取锁就是acquire(1),释放锁就是release(1)
下面我们模拟之前提到的线程永远无法获取锁的情况
按照我们的分析，是有可能存在线程永远阻塞情况，但事实却没有这个情况，什么原因呢？奥妙就在LockSupport.unpark()这个方法上
/**
* Makes available the permit for the given thread, if it
* was not already available. If the thread was blocked on
* {@code park} then it will unblock. Otherwise, its next call
* to {@code park} is guaranteed not to block. This operation
* is not guaranteed to have any effect at all if the given
* thread has not been started.
*
* @param thread the thread to unpark, or {@code null}, in which case
* this operation has no effect
*/
public static void unpark(Thread thread) {
if (thread != null)
UNSAFE.unpark(thread);
}
注释很明确的说明了原因，假如一个线程没有被阻塞时调用unpark方法，再调用park方法，该线程是不会被阻塞的。这样上面的问题也就解释通了。
结语
AbstractQueuedSynchronizer 本次就分析到这里，我们只分析了最核心的两个方法acquire()和release()两个方法，涉及到的Node状态只有0，SIGNAL(-1), 其他方法的分析之后解决具体的实现类再细说。