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 (;;) {
springboot其实就是springfinal Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
< = 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)) {
< = 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)) {
< = node;
return t;
}
}
}
}
同样tail 为 null , 进⼊到 compareAndSetHead 。
compareAndSetHead(new Node())
/**
* CAS head field. Used only by enq.
*/
private final boolean compareAndSetHead(Node update) {
return unsafepareAndSwapObject(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);
< = 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);
< = 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 unsafepareAndSwapInt(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 标记后继结点的线程状态。
以上就是本⽂的全部内容,希望对⼤家的学习有所帮助,也希望⼤家多多⽀持。

版权声明:本站内容均来自互联网,仅供演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198,我们将在24小时内删除。