CountDownLatch源码解析之await()
countdownlatch 源码解析—— await(),具体内容如下
说了一下countdownlatch的使用方法。这篇文章就从源码层面说一下await() 的原理。
我们已经知道await 能够让当前线程处于阻塞状态,直到锁存器计数为零(或者线程中断)。
下面是它的源码。
end.await(); ↓ public void await() throws interruptedexception { sync.acquiresharedinterruptibly(1); }
sync 是countdownlatch的内部类。下面是它的定义。
private static final class sync extends abstractqueuedsynchronizer { ... }
它继承了abstractqueuedsynchronizer。abstractqueuedsynchronizer 这个类在java线程中属于一个非常重要的类。
它提供了一个框架来实现阻塞锁,以及依赖fifo等待队列的相关同步器(比如信号、事件等)。
继续走下去,就跳到 abstractqueuedsynchronizer 这个类中。
sync.acquiresharedinterruptibly(1); ↓ public final void acquiresharedinterruptibly(int arg) //abstractqueuedsynchronizer throws interruptedexception { if (thread.interrupted()) throw new interruptedexception(); if (tryacquireshared(arg) < 0) doacquiresharedinterruptibly(arg); }
这里有两个判断,首先判断线程是否中断,然后再进行下一个判断,这里我们主要看看第二个判断。
protected int tryacquireshared(int acquires) { return (getstate() == 0) ? 1 : -1; }
需要注意的是 tryacquireshared 这个方法是在sync 中实现的。
abstractqueuedsynchronizer 中虽然也有对它的实现,但是默认的实现是抛一个异常。
tryacquireshared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。
我们可以看到sync 中是通过判断state 是否为0 来返回对应的 int 值的。
那么 state 又代表什么?
/** * the synchronization state. */ private volatile int state;
上面代码很清楚的表明 state 是表示同步的状态 。
需要注意的是 state 使用 volatile 关键字修饰。
volatile 关键字能够保证 state 的修改立即被更新到主存,当有其他线程需要读取时,会去内存中读取新值。
也就是保证了state的可见性。是最新的数据。
走到这里 state 是多少呢?
这里我们就需要看一看countdownlatch 的 构造函数了。
countdownlatch end = new countdownlatch(2); ↓ public countdownlatch(int count) { if (count < 0) throw new illegalargumentexception("count < 0"); this.sync = new sync(count); } ↓ sync(int count) { setstate(count); }
原来构造函数中的数字就是这个作用啊,用来set state 。
所以我们这里state == 2 了。tryacquireshared 就返回 -1。进入到下面
doacquiresharedinterruptibly(arg); ↓ private void doacquiresharedinterruptibly(int arg) throws interruptedexception { final node node = addwaiter(node.shared); boolean failed = true; try { for (;;) { final node p = node.predecessor(); if (p == head) { int r = tryacquireshared(arg); if (r >= 0) { setheadandpropagate(node, r); p.next = null; // help gc failed = false; return; } } if (shouldparkafterfailedacquire(p, node) && parkandcheckinterrupt()) throw new interruptedexception(); } } finally { if (failed) cancelacquire(node); } }
ok,这段代码有点长,里面还调用了几个函数。我们一行一行的看。
第一行 出现了一个新的类 node。
node 是aqs(abstractqueuedsynchronizer)类中的内部类,定义了一种链式结构。如下所示。
+------+ prev +-----+ +-----+ head | | <---- | | <---- | | tail +------+ +-----+ +-----+
千万记住这个结构。
第一行代码中还有一个方法 addwaiter(node.shared) 。
addwaiter(node.shared) //node.shared 表示该结点处于共享模式 ↓ private node addwaiter(node mode) { node node = new node(thread.currentthread(), mode); // try the fast path of enq; backup to full enq on failure node pred = tail; // private transient volatile node tail; if (pred != null) { node.prev = pred; if (compareandsettail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }
首先是构造了一个node,将当前的线程存进去了,模式是共享模式。
tail 表示 这个等待队列的队尾,此刻是null. 所以 pred == null ,进入到enq(node) ;
enq(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; } } } }
同样tail 为 null , 进入到 compareandsethead 。
compareandsethead(new node()) ↓ /** * cas head field. used only by enq. */ private final boolean compareandsethead(node update) { return unsafe.compareandswapobject(this, headoffset, null, update); }
这是一个cas操作,如果head 是 null 的话,等待队列的 head 就会被设置为 update 的值,也就是一个新的结点。
tail = head; 那么此时 tail 也不再是null了。进入下一次的循环。
这次首先将node 的 prev 指针指向 tail ,然后通过一个cas 操作将node 设置为尾部,并返回了队列的 tail ,也就是 node 。
等待队列的模型变化如下
+------+ prev +----------------+ head(tail) | | <---- node | currentthread | +------+ +----------------+ ↓ +------+ prev +----------------+ head | | <---- node(tail) | currentthread | +------+ +----------------+
ok,到了这里await 方法 就返回了,是一个 thread 等于当前线程的node。
返回到 doacquiresharedinterruptibly(int arg) 中,进入下面循环。
for (;;) { final node p = node.predecessor(); if (p == head) { int r = tryacquireshared(arg); if (r >= 0) { setheadandpropagate(node, r); p.next = null; // help gc failed = false; return; } } if (shouldparkafterfailedacquire(p, node) && parkandcheckinterrupt()) throw new interruptedexception(); }
这个时候假设state 仍然大于0,那么此时 r < 0,所以进入到 shouldparkafterfailedacquire 这个方法 。
shouldparkafterfailedacquire(p, node) ↓ private static boolean shouldparkafterfailedacquire(node pred, node node) { int ws = pred.waitstatus; if (ws == node.signal) //static final int signal = -1; /* * this node has already set status asking a release * to signal it, so it can safely park. */ return true; if (ws > 0) { /* * predecessor was cancelled. skip over predecessors and * indicate retry. */ 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. */ compareandsetwaitstatus(pred, ws, node.signal); } return false; } ↓ /** * cas waitstatus field of a node. */ private static final boolean compareandsetwaitstatus(node node, int expect, int update) { return unsafe.compareandswapint(node, waitstatusoffset, expect, update); }
可以看到 shouldparkafterfailedacquire 也是一路走,走到 compareandsetwaitstatus。
compareandsetwaitstatus 将 prev 的 waitstatus 设置为 node.signal 。
node.signal 表示后续结点中的线程需要被unparking(类似被唤醒的意思)。该方法返回false。
经过这轮循环,队列模型变成下面状态
+--------------------------+ prev +------------------+ head | waitstatus = node.signal | <---- node(tail) | currentthread | +--------------------------+ +------------------+
因为shouldparkafterfailedacquire返回的是false,所以后面这个条件就不再看了。继续 for (;;) 中的循环。
如果state仍然大于0,再次进入到 shouldparkafterfailedacquire。
这次因为head 中的waitstatus 为 node.signal ,所以 shouldparkafterfailedacquire 返回true。
这次就需要看parkandcheckinterrupt 这个方法了。
private final boolean parkandcheckinterrupt() { locksupport.park(this); return thread.interrupted(); }
ok,线程没有被中断,所以,返回false。继续 for (;;) 中的循环。
如果state 一直大于0,并且线程一直未被中断,那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。
那么什么情况下跳出循环呢?也就是什么情况下state 会 小于0呢? 下一篇文章 我将说明。
总结一下,await() 方法 其实就是初始化一个队列,将需要等待的线程(state > 0)加入一个队列中,并用waitstatus 标记后继结点的线程状态。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
下一篇: ae导出视频出错提示设置不匹配怎么办?
推荐阅读
-
CountDownLatch源码解析之await()
-
CountDownLatch源码解析之countDown()
-
Spring5源码解析4-refresh方法之invokeBeanFactoryPostProcessors
-
Laravel框架源码解析之模型Model原理与用法解析
-
Spring源码解析之ConfigurableApplicationContext
-
angularjs 源码解析之injector
-
angularjs 源码解析之scope
-
Java并发之ReentrantLock类源码解析
-
Java并发系列之CountDownLatch源码分析
-
Android图片加载利器之Picasso源码解析