以下以ReentrantLock的NofaiySync分析AQS(又或者说CLH变种)的实现方式。
1、尝试获取资源
调用NofairSync的lock方法的时候,会先尝试对资源state加锁,失败的时候还会尝试获取锁。
final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}
2、获取失败加入队列
这样符合一般的资源竞争策略,AQS内部参照了CLH维护了一个由资源竞争者(线程)所创立的队列 -> 失败则加入队列。
public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}
这里即使是加入队列,也不会忘记第一步尝试获取资源,目的一是如果资源能够快速释放,那么可以不入队列就可以获取,毕竟这里本身就是一个非公平锁的实现。然后如果依然获取失败,则会把当前节点包装成一个独占节点(这里节点类型看具体的实现而异,例如ReentrantLock是独占锁,所以节点都是Node.EXCLUSIVE)。
private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;}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;}}}}
这里addWaiter和enq都是为了使当前节点能够成功进入队列所执行操作,每一个AQS的节点,都是把当前想尝试获取资源的线程Thread封装成了一个节点,并不断尝试加入到队列的尾节点。当然从enq的实现我们可以看到队列不是一开始就存在的,只有当线程获取资源失败并且需要入队时,才开始创建,这相当于是一个惰性初始化。
3、一个虚结点和两段真空期
除此之外,可以看到队列的头节点是一个虚结点,这个很关键,AQS的头节点都是虚结点。
- 在初始化的时候,会构造一个null节点放在头。
- 当有线程占有锁的时候,会把上一个为null的节点移除,同时将自己置为null,作为新的头,而此时,自己已经上位了,获取到了锁
- 所以始终会有一个null节点放在队列头
这里把AQS的头节点设置为伪结点的原因其中一个:
/*** Sets head of queue to be node, thus dequeuing. Called only by* acquire methods. Also nulls out unused fields for sake of GC* and to suppress unnecessary signals and traversals.** @param node the node*/private void setHead(Node node) {head = node;node.thread = null;node.prev = null;}
为了方便GC回收。
这里除了头节点是虚结点之外,还有两个地方值得注意。加入whead和wtail表示队列的头尾两个节点,刚开始时两个节点都是null,直到队列创建,CAS一个虚结点为头节点,再把whead=wtail。这里存在一个真空期,就是队列的whead设置为一个虚结点和将wtail指向同一个节点。所以一般AQS判断队列是否生成,基本都是用whead == null 来判断的。
enq还有一个地方值的注意的是,for死循环第二个else域,新的节点node是先将prev指针域指向wtail(t),再compareAndSetTail(t, node),最后再把之前t的next指针设置为node。这里的CAS设置尾节点,以及t.next = node也同样存在一个真空期。这样会导致如果遍历为头节点,从next域开始遍历,会遍历不到当前新的尾节点。所以后续看到AQS的遍历,一般是从尾节点用prev指针域开始遍历。
4、入队成功后挂起线程
为了避免线程不断抢占CPU资源,一般AQS在将线程加入队列成功后,就会调用LockSupport.park将线程挂起。具体在acquireQueue这个方法里面实现。
final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}
如果当前节点的前驱节点predecessor为头节点,那么尝试获取资源,这个是for死循环内第一个if的内容。为了方便GC,同样也把predecessor的next指针域设置为null,failed失败标志变量为false。所以说在AQS内部,资源被释放之后,会交给头节点的下一个节点去获取。(NofairSync是在自身的实现做到抢夺的,和AQS内部无关)
关键是第二个if语句
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;if (ws == Node.SIGNAL)return true;if (ws > 0) {do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}
5、SIGNAL为什么表示的是释放资源时唤醒下一个节点
shouldParkAfterFailedAcquire 这里面由三个条件判断分支:
当前驱节点pred的状态为SIGNAL的时候,直接放回true,为什么我当前线程判断是否要挂起需要判断前驱节点的状态是否为SIGNAL?因为SIGNAL的状态表示,如果我释放资源的时候,如果当前我的状态为SIGNAL时,需要把下一个节点唤醒。换句话说,SGINAL就是为了下一个节点能够被成功唤醒做准备的。这里我们需要回顾AQS的release释放资源的方法
public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}
头节点释放资源时,需要判断head是否为空,前面说了这个时AQS用来判断队列是否被创建的关键,其次还判断了h.waitStatus是否为0,众所周知,刚刚初始化的节点,waitStatus为0,不为0则表示有后继节点修改了该值。这里我曾经想过是否可以判断next指针是否为空来判断是否有后续节点。虽然上面提到的第二个真空期的时候,后面的新节点node是先CAS为wtail,再把前一个wtail(这种情况应该是whead=wtail,也就是头节点和尾节点指向同一个)的next域设置为新的节点。如果刚好在这个真空期头节点释放了资源,就会因为没有找到next指针域,而直接返回了release方法。但即使这样,再enq方法中,新的节点node依然加入队列成功,虽然whead释放了资源,但是队列的head依然指向于他,所以新的节点依然在acquireQueued的for死循环的第一个if语句获取资源成功。把whead指向于新的节点,并把之前的whead和新节点的之间联系next指针设置为null。
这样看起来似乎也是可以走的通,再考虑竞争资源的还有其他的线程,如果此时过来了一个新的竞争者node1,在node还未在acquireQueued获取到资源的时候,率先在方法lock的第一个if判断条件获取到资源了,会怎样?这里为了方便解说在两个方法同时拿出来
node也就是原来whead的下一个节点,此时刚进入acquireQueue方法
final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}
}node1新来的竞争者,调用了lock方法,刷新在第一个if条件中CAS获取资源成功
final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);
}
这样会导致没有在队列中的节点node1先获取了资源,看起来也没什么问题,后续node1释放资源后,依然会去队列释放下一个节点node。
所以这样看起来,判断next指针和使用waitStatus判断没有区别,那为什么最终AQS采用的是waitStatus,回顾一下,当enq方法成功CAS将当前节点node设置为尾节点,并且把前一个节点的next设置node,到之后设置前一个节点的waitStatus为SIGNAL,这里,走了哪些逻辑,也就是acquireQueue的实现是什么。
final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}
设置waitStatus为SGINAL的地方在shouldParkAfterFailedAcquire,不过在此之前,tryAcquire可能会抛出异常,我们看下tryAcquire的实现
protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}
这里如果nextc值溢出了,会抛出Maximun lock count exceeded的错误,所以仅凭next指针是无法保证下一个节点有效。而head的waitStatus如果发生了变化,说明已经正确走过这一步,并且在shouldParkFailedAcquire设置了waitStatus的状态,至于是什么状态就需要到unparkSuccessors判断。
所以这里也就是说明为什么需要用waitStatus判断下一个节点是否准备就绪的做法。其实还有一个原因(个人处于非代码的考虑),因为next在整个AQS中都是不可信的,所以无论在任何时候,做任何判断,都不会用到next指针,next指针只是作为一个遍历。
6、为什么waitStatus=SIGNAL不放在当前节点内部,而是前驱节点中
还是那个原因,next指针时不可信的,如果我当前节点head释放了资源,那么需要去查看下一个节点是否准备就绪就要用到next指针,有可能此时拿到的是一个null,因为CAS和设置next指针存在真空期(enq方法)。所以自然的也需要把waitStatus=SIGNAL设置在前驱节点,也自然需要一个dummy的头节点,这就是除了GC外需要一个伪结点的原因。
7、acquireQueued中interrupted的作用
这个比较简单,因为LockSupport.park如果被唤醒,是不会有返回值给出原因的,所以需要拿当前线程的状态去Thread.interrputed()获取到是否中断。最终把线程挂起,直到资源可用,头节点把下一个节点unpark后尝试获取资源成功。
private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}
8、引用文章
1、Reentrantlock源码--深入理解AQS
2、Java多线程 20 - AbstractQueuedSynchronizer详解(1)
3、Java AQS源码解读
9、思考
A、acquireQueued 方法里,为什么还要再tryAcquire?
以独占模式来说,对于这个问题,我是这么想的:
时刻1:线程B尝试获取锁,但是,由于锁被线程A持有,所以,线程B准备调用addWaiter,将自己入到队列(但还没有和head节点产生指针连接)
时刻1:同一时刻,线程A尝试释放锁,进入release方法,调用子类的tryRelease(),将代表锁持有次数的state置为0(代表锁没有被任何线程持有),进入unparkSuccessor方法,发现并没有后继节点(因为新节点还未入队),所以不会唤醒任何线程,到这里,线程A释放锁操作完成。
时刻2:线程B调用addWaiter方法完毕,已经入队,并和head节点产生指针连接
时刻3:线程B调用acquireQueued方法(如下方代码展示),如果在这个方法里面不调用tryAcquire,就会发生这样的情况:明明可以获取锁,但是线程却被休眠了,进而导致整个同步队列不可用
所以,再次调用tryAcquire是为了防止新节点还未入队,但是头结点已经释放了锁,导致整个同步队列瘫痪的情况发生。
final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}
B、parkAndCheckInterrupt会在线程中断时返回,但是因为acquireQueued是死循环,又会进去一次导致unpark?
是的,如果需要ReentrantLock响应中断,那么就不应该调用lock,而是lockInterruptibly(),这个可以响应中断,并抛出异常。具体看下实现:
public void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);}public final void acquireInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (!tryAcquire(arg))doAcquireInterruptibly(arg);}
acquireInterruptibly前面部分都是为了判断线程是否中断过,避免再次去竞争资源就可以直接返回,所以我们可以直接看到尝试获取资源tryAcquire() 失败后的 doAcquireInterrputibly()方法。
private void doAcquireInterruptibly(int arg)throws InterruptedException {final Node node = addWaiter(Node.EXCLUSIVE);boolean failed = true;try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}}
因为获取失败,和前面的方法一样也是调用addWaiter加入队列的尾部。不一样的是在死循环的第二个if判断,如果方法从parkAndCheckInterrupt中返回并且为true,说明Thread.isInterrupted()返回true,线程被中断了,那么直接抛出异常,不像acquireQueued中是把一个interrupted变量设置为ture,这样就可以从死循环中退出。
C、为什么需要虚结点?
其实总结整篇文章,可以看到虚结点就是当前拿到资源的节点。只不过线程域被设置为了null。设置为null目的是为了方便GC,这里的疑问应该是为何拿到资源的节点,并且指针域已经为null还会留在队列里面,目的是跟前面同样提到的waitStatus=SIGNAL有关。 既然需要每个节点都需要把前一个节点的waitStatus=SIGNAL保证自身能够被唤醒,那么如果把虚结点去掉会导致队列头没有办法设置waitStatus=SIGNAL。
所以问题又回到为什么SIGNAL表示资源释放时需要唤醒下一个节点,以及为什么不把这个信息放到自身节点上面。
再次总结下头节点状态为SIGNAL时,需要释放资源时需要唤醒下一个节点。按照正常的链表逻辑,我只需要判断下一个节点next是否为空即可,但是回顾acquireQueued这个方法,不难发现在死循环的第一个if判断可能会抛出Error,导致即使头节点的下一个指针域next有值,但也有该节点本身是有异常的,但是头节点却无法得知,头节点只是被动被人设置了next指针。所以最好还是把waitStatus来判断,因为把waitStatus设置为SIGNAL在shouldParkAfterFailedAcquire()这个方法中获得,修改完就是挂起了,这样表示下一个节点已经准备好了,可以随时恢复。至于为什么不把自己的waitStatus设置为SIGNAL,让头节点去判断下一个节点的waitStatus为SIGNAL呢,答案已经很明显啦,我不能用next指针获取下一个节点的信息,如果去判断下一个节点是否为SIGNAL就必须使用next指针,所以还是让下一个节点把前一个节点状态设置为SIGNAL,表示我已经准备好了。再加上一点最重要的,尽量避免遍历整个队列情况出现,因为这个是在并发的情况下。
还有一点,为什么acquireQueued中需要tryAcquire,如果没有这个就不会报错,那么就可以用链表判断下一个是否为null了。的确如此,关于这个可以看到思考A有结论,主要是因为多个操作就不是原子的,防止头节点释放资源,但是新的节点刚好在enq只是刚好创建了节点封装了线程,跑到acquireQueued挂起后没有被唤醒。