一篇帶給你CountDownLatch實(shí)現(xiàn)原理
前言
CountDownLatch是多線程中一個(gè)比較重要的概念，它可以使得一個(gè)或多個(gè)線程等待其他線程執(zhí)行完畢之后再執(zhí)行。它內(nèi)部有一個(gè)計(jì)數(shù)器和一個(gè)阻塞隊(duì)列，每當(dāng)一個(gè)線程調(diào)用countDown()方法后，計(jì)數(shù)器的值減少1。當(dāng)計(jì)數(shù)器的值不為0時(shí)，調(diào)用await()方法的線程將會(huì)被加入到阻塞隊(duì)列，一直阻塞到計(jì)數(shù)器的值為0。
常用方法
- public class CountDownLatch {
- //構(gòu)造一個(gè)值為count的計(jì)數(shù)器
- public CountDownLatch(int count);
- //阻塞當(dāng)前線程直到計(jì)數(shù)器為0
- public void await() throws InterruptedException;
- //在單位為unit的timeout時(shí)間之內(nèi)阻塞當(dāng)前線程
- public boolean await(long timeout, TimeUnit unit);
- //將計(jì)數(shù)器的值減1，當(dāng)計(jì)數(shù)器的值為0時(shí)，阻塞隊(duì)列內(nèi)的線程才可以運(yùn)行
- public void countDown();
- }
下面給一個(gè)簡(jiǎn)單的示例：
- package com.yang.testCountDownLatch;
- import java.util.concurrent.CountDownLatch;
- public class Main {
- private static final int NUM = 3;
- public static void main(String[] args) throws InterruptedException {
- CountDownLatch latch = new CountDownLatch(NUM);
- for (int i = 0; i < NUM; i++) {
- new Thread(() -> {
- try {
- Thread.sleep(2000);
- System.out.println(Thread.currentThread().getName() + "運(yùn)行完畢");
- } catch (InterruptedException e) {
- e.printStackTrace();
- } finally {
- latch.countDown();
- }
- }).start();
- }
- latch.await();
- System.out.println("主線程運(yùn)行完畢");
- }
- }
輸出如下：
看得出來(lái)，主線程會(huì)等到3個(gè)子線程執(zhí)行完畢才會(huì)執(zhí)行。
原理解析
類圖
可以看得出來(lái)，CountDownLatch里面有一個(gè)繼承AQS的內(nèi)部類Sync，其實(shí)是AQS來(lái)支持CountDownLatch的各項(xiàng)操作的。
CountDownLatch(int count)
new CountDownLatch(int count)用來(lái)創(chuàng)建一個(gè)AQS同步隊(duì)列，并將計(jì)數(shù)器的值賦給了AQS的state。
- public CountDownLatch(int count) {
- if (count < 0) throw new IllegalArgumentException("count < 0");
- this.sync = new Sync(count);
- }
- private static final class Sync extends AbstractQueuedSynchronizer {
- Sync(int count) {
- setState(count);
- }
- }
countDown()
countDown()方法會(huì)對(duì)計(jì)數(shù)器進(jìn)行減1的操作，當(dāng)計(jì)數(shù)器值為0時(shí)，將會(huì)喚醒在阻塞隊(duì)列中等待的所有線程。其內(nèi)部調(diào)用了Sync的releaseShared(1)方法
- public void countDown() {
- sync.releaseShared(1);
- }
- public final boolean releaseShared(int arg) {
- if (tryReleaseShared(arg)) {
- //此時(shí)計(jì)數(shù)器的值為0，喚醒所有被阻塞的線程
- doReleaseShared();
- return true;
- }
- return false;
- }
tryReleaseShared(arg)內(nèi)部使用了自旋+CAS操將計(jì)數(shù)器的值減1，當(dāng)減為0時(shí)，方法返回true，將會(huì)調(diào)用doReleaseShared()方法。對(duì)CAS機(jī)制不了解的同學(xué)，可以先參考我的另外一篇文章淺探CAS實(shí)現(xiàn)原理
- protected boolean tryReleaseShared(int releases) {
- //自旋
- for (;;) {
- int c = getState();
- if (c == 0)
- //此時(shí)計(jì)數(shù)器的值已經(jīng)為0了，其他線程早就執(zhí)行完畢了，當(dāng)前線程也已經(jīng)再執(zhí)行了，不需要再次喚醒了
- return false;
- int nextc = c-1;
- //使用CAS機(jī)制，將state的值變?yōu)閟tate-1
- if (compareAndSetState(c, nextc))
- return nextc == 0;
- }
- }
doReleaseShared()是AQS中的方法，該方法會(huì)喚醒隊(duì)列中所有被阻塞的線程。
- private void doReleaseShared() {
- for (;;) {
- Node h = head;
- if (h != null && h != tail) {
- int ws = h.waitStatus;
- if (ws == Node.SIGNAL) {
- if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
- continue; // loop to recheck cases
- unparkSuccessor(h);
- }
- else if (ws == 0 &&
- !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
- continue; // loop on failed CAS
- }
- if (h == head) // loop if head changed
- break;
- }
- }
這段方法比較難理解，會(huì)另外篇幅介紹。這里只要認(rèn)為該段方法會(huì)喚醒所有因調(diào)用await()方法而阻塞的線程。
await()
當(dāng)計(jì)數(shù)器的值不為0時(shí)，該方法會(huì)將當(dāng)前線程加入到阻塞隊(duì)列中，并把當(dāng)前線程掛起。
- public void await() throws InterruptedException {
- sync.acquireSharedInterruptibly(1);
- }
同樣是委托內(nèi)部類Sync，調(diào)用其
acquireSharedInterruptibly()方法
- public final void acquireSharedInterruptibly(int arg)
- throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- if (tryAcquireShared(arg) < 0)
- doAcquireSharedInterruptibly(arg);
- }
接著看Sync內(nèi)的tryAcquireShared()方法，如果當(dāng)前計(jì)數(shù)器的值為0，則返回1，最終將導(dǎo)致await()不會(huì)將線程阻塞。如果當(dāng)前計(jì)數(shù)器的值不為0，則返回-1。
- protected int tryAcquireShared(int acquires) {
- return (getState() == 0) ? 1 : -1;
- }
tryAcquireShared方法返回一個(gè)負(fù)值時(shí)，將會(huì)調(diào)用AQS中的
doAcquireSharedInterruptibly()方法，將調(diào)用await()方法的線程加入到阻塞隊(duì)列中，并將此線程掛起。
- private void doAcquireSharedInterruptibly(int arg)
- throws InterruptedException {
- //將當(dāng)前線程構(gòu)造成一個(gè)共享模式的節(jié)點(diǎn)，并加入到阻塞隊(duì)列中
- 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);
- }
- }
同樣，以上的代碼位于AQS中，在沒(méi)有了解AQS結(jié)構(gòu)的情況下去理解上述代碼，有些困難，關(guān)于AQS源碼，會(huì)另開(kāi)篇幅介紹。
使用場(chǎng)景
CountDownLatch的使用場(chǎng)景很廣泛，一般用于分頭做某些事，再匯總的情景。例如：
數(shù)據(jù)報(bào)表：當(dāng)前的微服務(wù)架構(gòu)十分流行，大多數(shù)項(xiàng)目都會(huì)被拆成若干的子服務(wù)，那么報(bào)表服務(wù)在進(jìn)行統(tǒng)計(jì)時(shí)，需要向各個(gè)服務(wù)抽取數(shù)據(jù)。此時(shí)可以創(chuàng)建與服務(wù)數(shù)相同的線程數(shù)，交由線程池處理，每個(gè)線程去對(duì)應(yīng)服務(wù)中抽取數(shù)據(jù)，注意需要在finally語(yǔ)句塊中進(jìn)行countDown()操作。主線程調(diào)用await()阻塞，直到所有數(shù)據(jù)抽取成功，最后主線程再進(jìn)行對(duì)數(shù)據(jù)的過(guò)濾組裝等，形成直觀的報(bào)表。
風(fēng)險(xiǎn)評(píng)估：客戶端的一個(gè)同步請(qǐng)求查詢用戶的風(fēng)險(xiǎn)等級(jí)，服務(wù)端收到請(qǐng)求后會(huì)請(qǐng)求多個(gè)子系統(tǒng)獲取數(shù)據(jù)，然后使用風(fēng)險(xiǎn)評(píng)估規(guī)則模型進(jìn)行風(fēng)險(xiǎn)評(píng)估。如果使用單線程去完成這些操作，這個(gè)同步請(qǐng)求超時(shí)的可能性會(huì)很大，因?yàn)榉?wù)端請(qǐng)求多個(gè)子系統(tǒng)是依次排隊(duì)的，請(qǐng)求子系統(tǒng)獲取數(shù)據(jù)的時(shí)間是線性累加的。此時(shí)可以使用CountDownLatch，讓多個(gè)線程并發(fā)請(qǐng)求多個(gè)子系統(tǒng)，當(dāng)獲取到多個(gè)子系統(tǒng)數(shù)據(jù)之后，再進(jìn)行風(fēng)險(xiǎn)評(píng)估，這樣請(qǐng)求子系統(tǒng)獲取數(shù)據(jù)的時(shí)間就等于最耗時(shí)的那個(gè)請(qǐng)求的時(shí)間，可以大大減少處理時(shí)間。
