CountDownLatch源码理解

CountDownLatch源码理解

CountDownLatch

await()

public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}

accquireSharedInterryptibly()

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {//如果线程中断，抛出异常if (Thread.interrupted())throw new InterruptedException();//判断latch是否为0，为0返回1，否则返回-1//返回1，说明线程不需要等待，执行业务逻辑//返回-1.说明线程需要入队 被挂起if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);
}

tryAcquireShared()

protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;
}

doAcquireSharedInterruptibly()

private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {//将当前线程加入AQS阻塞队列，并且将状态设置为SHAREDfinal Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {//拿到当前线程的前驱节点final Node p = node.predecessor();//如果前驱节点是头节点，并且当前latch为0，就执行 setHeadAndPropagate(),并且帮助老的头节点出队if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}//执行到这里说明，前驱节点不是头节点  或者前驱节点是头节点但是latch不为0 就需要将当前线程挂起if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}
}

await()里面的唤醒逻辑

setHeadAndPropagate()

private void setHeadAndPropagate(Node node, int propagate) {Node h = head; // Record old head for check below//将当前节点设置为头节点setHead(node);//propagate必为1if (propagate > 0 || h == null || h.waitStatus < 0 ||(h = head) == null || h.waitStatus < 0) {//得到当前节点的后继节点Node s = ;if (s == null || s.isShared())//如果当前节点是尾节点 或者当前节点的后继节点是SIGNALdoReleaseShared();}
}

doReleaseShared()

//1.当前线程为将latch修改为0的线程，会进入到这里去唤醒线程(会给break掉,走从头节点开始唤醒的逻辑)
//2.在dASI方法里被park起的线程被唤醒后经过自旋也会走到这里
private void doReleaseShared() {for (;;) {Node h = head;if (h != null && h != tail) {//执行到这里，说明阻塞队列中除了头节点一定还有其他节点int ws = h.waitStatus;if (ws == Node.SIGNAL) {//为什么用cas？//因为可能老的头节点还没有跳出循环，会和当前头节点去竞争唤醒后继线程//竞争失败就继续唤醒下一个，感觉这个会提高吞吐量if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue;            // loop to recheck casesunparkSuccessor(h);}else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS}//条件成立：//1.说明刚刚唤醒的 后继节点，还没执行到 setHeadAndPropagate方法里面的 设置当前唤醒节点为head的逻辑。//这个时候，当前线程 直接跳出去...结束了..//此时用不用担心，唤醒逻辑 在这里断掉呢？、//不需要担心，因为被唤醒的线程 早晚会执行到doReleaseShared方法。//2.h == null  latch创建出来后，没有任何线程调用过 await() 方法之前，//有线程调用untDown()操作 且触发了 唤醒阻塞节点的逻辑..//3.h == tail  -> head 和 tail 指向的是同一个node对象//条件不成立：//被唤醒的节点 非常积极，直接将自己设置为了新的head，此时 唤醒它的节点（前驱），执行h == head 条件会不成立..//此时 head节点的前驱，不会跳出 doReleaseShared 方法，会继续唤醒 新head 节点的后继...if (h == head)                   // loop if head changedbreak;}
}

CountDown()

public void countDown() {leaseShared(1);
}

tryReleaseShared()

protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}
}

releaseShared()

public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;
}