多线程之美5一 AbstractQueuedSynchronizer源码分析<一>
夕阳下飞奔的猪 人气:2AQS的源码分析 <一>
目录结构
1、什么是CAS ?
2、同步器类结构
3、CLH同步队列
4、AQS中静态内部类Node
5、方法分析
5.1、acquire(int arg )
5.2、release(int arg) 释放锁
6、总结
前言
在多线程环境下，我们一般会对临界区资源（共享资源）进行加锁,释放锁，保证同一时刻最多只有一个线程（独占模式），就如去公共厕所里，在使用一个小房间时会加锁避免自己在使用的时候，别人突然闯进来一样，引起不必要的麻烦，在使用完后，再打开锁，其他人才可使用；还有生产者消费者模型中，线程之间要同步，需要等待和通知机制，来协调线程合作。那么这些是这么实现的？如可重入锁ReentrantLock, 读写锁ReadWriteLock, 信号量 Semaphore, 计数器CountDownLatch，这些都会涉及线程之间的协调同步，那么会有一个抽象的结构，将这些需要共用的功能抽离出来，统一来满足要求吗？我们一起来看看AbstractQueuedSynchronizer 这个抽象类，如何来实现这些功能和其设计的巧妙, 我们能看到Doug lea 大佬在很多地方使用的循环CAS操作(自旋锁)。
1、什么是CAS ?
CAS 即 compare and swap 比较并交换， 涉及到三个参数，内存值V, 预期值A, 要修改的新值B, 拿着预期值A与内存值V比较，相等则符合预期，将内存值V更新为B, 否则不相等，不能更新V;
比如我想去果篮拿1个苹果，我预期篮子是5个，而果篮中实际有2个，被其他人早已偷偷吃了几个我还不知道！于是我的期望值5与果篮值实际2不符，那么我就不能更新篮子的苹果数量了。这和数据库中乐观锁机制一样。后面我想用一篇文章单独总结一下CAS, 如它存在的ABA问题以及加上版本号来解决ABA问题， 还有用CAS实现自旋锁等。
2、同步器类结构
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizerAbstractQueuedSynchronizer （以后简称AQS）是一个抽象类，定义了一些通用功能以及子类需要根据自身特性重写实现的方法，有两个内部类 Node和ConditionObject, 以及2个链表结构的队列，一个是双向链表的同步队列，用来存放尝试获取锁，却未获取到锁需等待的线程，另一个是单向链表的条件队列，用来存放某些线程已经获取到锁了，为了等待某些事件（如IO事件,mq消息等）主动放弃锁挂起等待条件的线程。
这里主要分析2个方法，分别是独占式获取锁 acquire, 释放锁release。
// 独占模式 EXCLUSIVE
acquire(int arg)
release(int arg)
// 共享模式 SHARED
acquireShared(int arg)
releaseShared(int arg)3、CLH同步队列
AQS 内部持有一个双向链表的队列（CLH队列，也称线程同步队列，用来存储尝试获取锁，未获取到等待的线程），AQS 持有head, tail节点，以及资源同步状态state 属性, 来表示有限的锁资源占用情况。
注：Head节点是一个空节点，是不关联线程的，因为Head节点表示当前已经占用资源在使用的线程，该线程已经不需要等待锁，在同步队列中作占位作用，它只是一个标志节点，其他线程才是对锁资源有需求，在队列中等待的，慢慢看来，后面代码会更有理解。
双向链表结构如下图：
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
/**
* 等待队列中的头节点，懒初始化，如果头节点存在，表示当前正在占用锁资源的线程。
*
*/
private transient volatile Node head;
/**
* 等待队列中尾节点
*/
private transient volatile Node tail;
/**
* 同步状态,在独占模式下 state=0,表示锁空闲，可获取,state =1,则表示被其他线程占用，需等待
* 就如火车上的厕所，绿灯表示可用，红灯表示里面有人，需要等待
*/
private volatile int state;4、AQS中静态内部类Node
CLH双向链表中每个node 中有关联的线程，节点状态信息，以及节点的前后指针，每个属性有volatile修饰，可保证多线程之间的可见性。
static final class Node {
// 获取锁的模式分为独占式和共享式
static final Node SHARED = new Node();
//独占式
static final Node EXCLUSIVE = null;
// 以下是节点在等待队列中的5种状态，初始为0，即waitStatus的值
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
// 以下4个属性分别用volatile修饰，保证多线程之间可见性
//节点的状态
volatile int waitStatus;
//前置节点
volatile Node prev;
// 后置节点
volatile Node next;
// 节点关联的线程
volatile Thread thread;Node中在队列中的等待状态 waitStatus,初始为0，还有以下4种状态和其含义：
- SIGNAL: waitStatus= -1,表示当前节点线程在使用完同步资源，释放资源后，需要去唤醒后面的一个节点中的线程。
- CANCELLED：waitStatus= 1，节点在循环尝试获取锁资源失败，被中断了，就不再需要再在队列中等待资源了，需要执行节点出队操作，将该节点移出该队列。就如一群小伙伴都在排队等待领取免费的辣条呢，都要排到小名了， 突然被妈妈叫着回去吃饭了，被强行中断了，那就要移出队列，让给其他人了哦。
- PROPAGATE：waitStatus= -3, 该状态是共享模式下，如CountDownLatch等实现同步时，可有多个线程同时获取锁资源，在独占模式下，锁资源有限，只有1个，一次最多有一个线程获取到锁资源，共享模式下，锁资源可有多个，可有多个线程同时获取到锁，如果锁资源数量可足够下一个节点中线程使用，会继续唤醒下一个。
- CONDITION：waitStatus= -2，表示在Condition条件等待队列中。
5、方法分析
5.1、acquire(int arg )
锁资源只有1个，同一时刻最多只有一个线程可获取到锁，独占式获取锁，其他未获取到锁的线程需在同步队列中等待。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
//如果在等待过程中，线程被标记中断了，响应中断
selfInterrupt();
}变形为：
public final void acquire(int arg) {
boolean ta = tryAcquire(arg)
if(!ta) {
Node aw = addWaiter(Node.EXCLUSIVE);
boolean isInterrupt = acquireQueued(aw,arg);
if(isInterrupt) {
//如果在等待过程中，线程被标记中断了，响应中断
selfInterrupt();
}
}
}1) tryAcquire(arg)： 尝试去获取锁资源，具体怎样获取锁由每个子类实现，AQS不做处理，统一抛出 throw new UnsupportedOperationException() 异常；
2) addWaiter(Node.EXCLUSIVE)：如果尝试获取锁失败，将该线程封装到一个新node节点中,添加到等待队列中
3) acquireQueued(node, arg) ：如果在head节点后，再次尝试去获取锁，可能占用锁的线程已经释放资源了，如果再次获取失败，在队列中寻找到安全位置（告知前置节点，使用完资源唤醒一下自己）后，则在队列中挂起，等待被唤醒，返回布尔类型，表示线程是否被中断。
下面来分别查看以上3个方法的源码：
1）tryAcquire() 是尝试去获取锁，需要每个具体实现类去实现，默认抛出异常
protected boolean tryAcquire(int arg) {
//抛出不支持操作异常
throw new UnsupportedOperationException();
}来看看 ReentrantLock 是怎样实现 tryAcquire() 去获取锁的，ReentrantLock 有公平方式和非公平两种方式，这里简单看看非公平方式的实现细节。
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}内部调用nonfairTryAcquire 方法,代码如下：
final boolean nonfairTryAcquire(int acquires) {
//1,获取当前线程对象
final Thread current = Thread.currentThread();
//2,获取同步状态state
int c = getState();
//3, c为0表示锁可用，cas更新为1,compareAndSetState内部有unsafe类 调用本地方法（native修饰）执行原子更新操作，多个线程同时进来更新，只会有一个更新成功。更新成功就设置当前线程对象，用来标识当前线程拥有锁了，是有主的，其他线程在我没释放锁之前不可占用，也是用来支持可重入的。
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
//4,如果当前线程等于当前已获取锁线程，可重入，同步状态 +1,返回true,允许再次获取锁
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}可重入锁 : 指线程在获取锁资源后，再次去获取同步锁，是允许再次获得的，不然存在产生死锁的风险，由上可知ReentrantLock 是支持可重入的，可重入多次，每重入一次 state +1，那么重入n 次，也需要释放n次（同理，每释放一次，state -1 ），不然state 无法恢复到到空闲状态0，导致其他线程无法获取到锁。
2）addWaiter(Node mode)，将当前线程封装为一个节点，添加到等待队列中。该方法分为2步操作，第一步利用CAS快速将节点入队，如果能入队成功，则返回，否则第二步执行 enq() 方法。
为什么？
为了执行效率和安全性两方面考虑，因在多线程环境下，同一时间可能并发执行多个线程，如果此时有2个线程都在执行入队操作，就有可能其中一个线程执行CAS 失败，他引用 的tail节点已经被更新，需要重新获取新的，此时就需要执行enq() 循环轮询去获取最新的tail节点执行入队操作，直到成功。
如图所示：
线程A在刚获取到pred = tail后，cpu调度切换到线程B，线程B此时也执行入队操作，入队成功，队列长度+1, tail更新，此时cpu调度 线程A，线程A再执行 compareAndSetTail(pred, node) 操作就会失败。
快速添加到队尾失败了，有两种情况，
1.是tail节点 为空，队列为空，那么需要初始化队列，添加一个空的head节点，表示当前已经获取锁资源，正在执行的线程。
2.是在cas添加到队尾后面时，有其他线程也在操作，抢先入队成功，导致自己获取的tail节点不是最新的，那就轮询获取最新的tail执行更新。
流程图如下：
addWaiter(Node mode) 代码如下：
private Node addWaiter(Node mode) {
/*1，创建 node节点,节点信息保存线程及模式
* Node.EXCLUSIVE for exclusive 独占模式
* Node.SHARED for shared 共享模式
*/
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
//2,如果队尾节点不为空，尝试将节点添加到 tail后面，建立双向链接
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//3,如果队尾节点为空，或者cas添加节点失败
enq(node);
return node;
}enq（node） 节点入队操作，for无限循环尝试去添加节点到队尾，直至成功。
private Node enq(final Node node) {
for (;;) {
//每新的一次循环，获取的是最新的tail，因tail节点是volatile修饰的，多线程之间内存可见，每次更新都会被刷新到内存
Node t = tail;
if (t == null) { // Must initialize
//1，head节点是一个 node对象，但是没有绑定线程，表示当前已经获取到锁资源，正在执行的线程，此时首尾相同，会再次走一遍循环，添加当前节点到head节点后面
if (compareAndSetHead(new Node()))
tail = head;
} else {
//2,轮询去设置节点到队尾
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}boolean acquireQueued(final Node node, int arg) 返回线程是否被中断
在上一步方法将节点添加到同步队列后，执行该方法的目的是：
如果node的前置节点是head, 再次尝试一次获取锁资源，如果已经被释放了，那(线程)就去获取锁执行。如果不是头节点或者尝试获取锁资源失败，那就在队列找个位置挂起等待了 。
此时需要找一个安全点，确保前面的有效节点（没被取消的节点）的线程执行后，能够通知唤醒自己，当前节点线程才安心挂起等待。
流程图如下：
源代码如下：
final boolean acquireQueued(final Node node, int arg) {
//默认是失败
boolean failed = true;
try {
//默认是没有被中断
boolean interrupted = false;
//无限循环
for (;;) {
final Node p = node.predecessor();
//1，如果前置节点是head节点，再去尝试获取锁，如果获取成功了，则将自己更新为head，此时当前线程可以执行了。
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
// 此时返回 false，没被中断
return interrupted;
}
//2,1失败或前置不是head,那么要定位一个有效的位置去阻塞等待，前面有些节点可能是被取消的，需要跳过这些节点，清除出队列，shouldParkAfterFailedAcquire就是来做这个事
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
//3,标记中断标记，返回被中断过
interrupted = true;
}
} finally {
// 4,正常情况下是一直会进行for循环，当跳出该循环，即出现异常，如被标记了中断，去调用阻塞，抛出中断异常,此时没有在队列中找到安全位置，因此将该节点移出队列
if (failed)
cancelAcquire(node);
}
}shouldParkAfterFailedAcquire 在失败获取锁资源后，寻找可以安全挂起的位置
如果前置节点被取消，一直向前找到节点 ws<=0 为止,注意蓝色箭头，指向的前后指针还在，但是node1对象已经没有被引用，下次gc会被回收，则出队。
流程图如下：
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
//1,前置节点已经是signal，在它释放资源后，会通知唤醒队列中下一个线程，所以当前位置就是安全阻塞等待点
return true;
if (ws > 0) {
// ws> 0，前置节点被取消
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
// 2,ws>0,那么就继续往前找，直到找到一个节点没被取消的,跳过了那些被取消的节点，这些节点后面会被GC
pred.next = node;
} else {
//3，尝试将前置节点 状态更新为signal,ws= -1, 进行下一次轮询
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}在找到安全位置后，挂起当前线程，等待被唤醒，如果下次被唤醒，首先检查一下自己是被前置节点唤醒 还是被中断唤醒的
private final boolean parkAndCheckInterrupt() {
// 挂起当前线程
LockSupport.park(this);
//当被唤醒，返回是否被中断标记位
return Thread.interrupted();
}最后finally中，线程被中断，出现异常，取消节点；cancelAcquire(Node node) 将该节点移出等待队列，线程都被中断了，就不再
需要去等待锁资源了。
出队有三种情况:
- 1, 节点在队尾 ,等于tail
- 2, 节点在中间位置，既不在head后面，也不是tail
注： 我理解这里最终实现是要达到第3步的效果，但是在cancelAcquire方法里并没有更新第2步蓝色箭头的前置指针，而是留给了其他线程在获取锁资源时，执行shouldParkAfterFailedAcquire方法查找安全位置挂起来实现的，这时 node 将没有再被引用，因此会被GC。
3, 被取消节点在head的后面，第二个节点
这里唯一做的就是去唤醒 node 的后继节点（如果为没被取消的节点），如果后继节点为空或者ws =1,那么就在队列从后向前遍历去唤醒一个有效节点，这个节点醒来做的事情是尝试去获取锁，并且将排在它队列前连续被取消的节点出队。所以这里更新节点的前后指针操作还是交给其他线程来做的，在 shouldParkAfterFailedAcquire方法实现的。
源代码如下：
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
//1,将节点与线程解绑
node.thread = null;
//2,跳过前面所有被取消的节点
// Skip cancelled predecessors
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
//3,获取前置节点后一个位置节点，不一定是 node,有可能是跳过的被取消的节点，留作cas更新用
Node predNext = pred.next;
//4,将当前节点状态改为取消 ws =1
node.waitStatus = Node.CANCELLED;
//5,第一种情况：尾节点，将node前置节点更新为tail,并断开前置节点的next后置指针，node将无引用，被gc
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
int ws;
// 第三种情况：node在中间位置，要做的就是让node前置节点 pred的next指针指向node后的没被取消的节点，跳过node
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
// 第二种情况，在head后面，那么要需向node后找没被取消节点，让head跳过node指向他，并唤醒该节点的线程。
unparkSuccessor(node);
}
node.next = node; // help GC
}
}唤醒后继者，该方法在release 方法也被执行，在线程使用完资源后，去唤醒下一个等待该资源的线程去执行。
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// 状态置为0,初始状态，也不让她通知下一个节点，此时节点应该是取消状态1，是大于0的，为啥还需再判断一次？，这个方法不只这里用，释放资源release也用，正常ws =-1,所以更新状态为0
//首先判断node后置节点，如果不行，再从尾向前遍历去找一个没被取消的节点 去唤醒其线程
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);
}5.2、release(int arg) 释放锁
public final boolean release(int arg) {
//1,释放资源前提是已经获取到
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
//2，head的waitStatus在独占模式应该是signal =-1
// 唤醒下一个节点，代码如上
unparkSuccessor(h);
return true;
}
return false;
}同样，tryRelease() 与 tryAcquire() 一样，尝试释放锁，AQS都未具体实现，抛出异常，留个子类具体去实现。
AQS中代码：
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}这里还是看一下ReentrantLock 的具体实现
protected final boolean tryRelease(int releases) {
//1,在没有可重入情况下，state =1, releases =1, 此时 c应该 =0
int c = getState() - releases;
//2,判断当前释放锁线程是否等于拿到锁的线程，不等就抛出异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
//3,可见只有c=0,情况才可成功释放锁
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}注：因ReentrantLock可重入的锁，由上代码可见，只有在 state=0的情况，才可成功释放锁，常见错误场景：在使用ReentrantLock lock n次，却没有unlock相应的n次，导致没有成功释放锁。
6、总结
1、 AQS的CLH队列的 pre指针可保证可靠性，next是保证不了的，看上面的代码分析，在遍历队列时，会向前遍历来查找安全点。
2、Condition 队列与 CLH同步队列两者的区别，同步队列中的所有线程是在等待获取锁的，Condition条件队列是某些之前已经获取到锁，因为要等待某个事件（如IO事件)或者与某个线程同步（等待另一线程执行结果），主动调用await方法，主动释放锁后，将该线程放入条件队列中的，所以在条件队列中线程不是等待获取锁，而是在等某个条件，下一篇我会来分析AQS的重要机制- 等待通知，靠其内部类ConditionObject实现，await和 signal来阻塞与 通知机制，可替代传统Object类提供的wait和 notify功能，而AQS可有多个条件来分类，比之应用更加灵活。
加载全部内容