• 欢迎访问搞代码网站,推荐使用最新版火狐浏览器和Chrome浏览器访问本网站!
  • 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏搞代码吧

CountDownLatch源码解析之await()

java 搞代码 4年前 (2022-01-05) 24次浏览 已收录 0个评论

这篇文章主要为大家详细解析了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.<strong style="color:transparent">来源gao@daima#com搞(%代@#码@网</strong> */ 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 标记后继结点的线程状态。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持gaodaima搞代码网

以上就是CountDownLatch源码解析之await()的详细内容,更多请关注gaodaima搞代码网其它相关文章!


搞代码网(gaodaima.com)提供的所有资源部分来自互联网,如果有侵犯您的版权或其他权益,请说明详细缘由并提供版权或权益证明然后发送到邮箱[email protected],我们会在看到邮件的第一时间内为您处理,或直接联系QQ:872152909。本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:CountDownLatch源码解析之await()

喜欢 (0)
[搞代码]
分享 (0)
发表我的评论
取消评论

表情 贴图 加粗 删除线 居中 斜体 签到

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址