13 - JUC CountDownLatch concurrent programming

CountDownLatch

什么是CountDownLatch

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

A CountDownLatch is initialized with a given count. The await methods block until the current count reaches zero due to invocations of the countDown() method, after which all waiting threads are released and any subsequent invocations of await return immediately. This is a one-shot phenomenon – the count cannot be reset. If you need a version that resets the count, consider using a CyclicBarrier.

A CountDownLatch is a versatile synchronization tool and can be used for a number of purposes. A CountDownLatch initialized with a count of one serves as a simple on/off latch, or gate: all threads invoking await wait at the gate until it is opened by a thread invoking countDown(). A CountDownLatch initialized to N can be used to make one thread wait until N threads have completed some action, or some action has been completed N times.

A useful property of a CountDownLatch is that it doesn’t require that threads calling countDown wait for the count to reach zero before proceeding, it simply prevents any thread from proceeding past an await until all threads could pass.

Sample usage: Here is a pair of classes in which a group of worker threads use two countdown latches:

• The first is a start signal that prevents any worker from proceeding until the driver is ready for them to proceed;
• The second is a completion signal that allows the driver to wait until all workers have completed.

上面这段话来自Oracle官方解释.

CountDownLatchIt is an auxiliary tool for thread synchronization,He can make one or more threads wait,until some operations in other threads complete,Then the previously waiting thread is released.His main idea is to initialize with a given counter,It then provides us with two core methods,await()和countDown()方法.通过调用await()方法去阻塞线程,until due to the callcountDown()method decrements the current counter by 0.Then all wereawait()threads are released,Of course this process is a one-off,If you want to implement repeat count,可以用 CyclicBarrier.基于以上的理解,We can get two basic usage scenarios for him

1. 初始化计数器为1,Simulate a simple opening and closing latch,Let all threads go throughawait()kept out of the latch,until a coordinating thread calls itcountDown()方法将计数器减1,Instantly all threads blocking the door can pass through the latch.
2. 初始化计数器为N,Make a thread wait outside the door,等NThreads complete their work to open the door,Let the waiting thread go through the latch.

Let's take a look at it with two examples.

应用场景

例1：School sports meet,The 100-meter race is underway.All the athletes are waiting at the starting line,As long as the referee whistle blows,Athletes can start playing immediately,Then wait for all athletes to finish the race,The referee can record the score,宣布比赛结束.This example can be abstracted into two multi-threading problems：

2. 有N个线程（运动员）Need to wait for a thread（裁判）give a signal,才可以开始工作
3. 有一个线程（裁判）Need to wait for the othersN个线程（运动员）both complete their respective jobs,Only then can he start his work

public class Referee {
    

    public static void main(String[] args) throws Exception {
    
        // Simulate the first problem
        CountDownLatch startSignal = new CountDownLatch(1);
        // Simulate the second problem
        CountDownLatch doneSignal = new CountDownLatch(6);

        // 1.1 我们创建6个线程,并调用start方法,The simulation has six athletes preparing to compete at the same time
        for (int i = 0; i < 6; i++) {
    
            new Thread(new Player(startSignal, doneSignal), "player " + i).start();
        }

        // The referee does some pre-match checks,检查OKWhistle afterward,比赛开始
        doReadyWork(startSignal);
        // Wait for all players to finish the game,记录分数
        recordScore(doneSignal);

    }

    private static void doReadyWork(CountDownLatch startSignal) {
    
        try {
    
            // The referee checks that everything is ready
            System.out.println("Ready..");
            Thread.sleep(3000);
            System.out.println("Go...");

            // 1.3 调用countDown()method to open the latch,The mock check is complete,吹口哨,All athletes begin
            startSignal.countDown();
        } catch (InterruptedException e) {
    
            e.printStackTrace();
        }
    }

    private static void recordScore(CountDownLatch doneSignal) {
    
        try {
    
            // 2.1 裁判通过await()method blocks this,Wait for the athlete over there to finish the game,将doneSignal的计数器减为0,He started to work
            doneSignal.await();
            System.out.println("Record the score...");
        } catch (InterruptedException e) {
    
            e.printStackTrace();
        }
    }
}

public class Player implements Runnable{
    
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;

    public Player(CountDownLatch startSignal, CountDownLatch doneSignal) {
    
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }

    @Override
    public void run() {
    
        try {
    
            // 1.2 调用await()method to block each thread,Simulate all the athletes waiting over there for the referee to whistle,When the whistle blows, the game begins
            startSignal.await();
            beginPlay();
            // 2.2 doneSignal的计数器为6,Each athlete ends the game,计数器减1,wait until reduced to0,Indicates that everyone has finished the game,
            // 那边通过doneSignalThe blocked thread can also start doing its thing
            doneSignal.countDown();
        } catch (InterruptedException e) {
    
            e.printStackTrace();
        }

    }

    public void beginPlay() {
    
        try {
    
            System.out.println(Thread.currentThread().getName() + " begin...");
            Thread.sleep(5000);
            System.out.println(Thread.currentThread().getName() + " complete...");
        } catch (InterruptedException e) {
    
            e.printStackTrace();
        }
    }
}

执行结果：
Ready..
Go...
player 1 begin...
player 0 begin...
player 2 begin...
player 4 begin...
player 3 begin...
player 5 begin...
player 2 complete...
player 0 complete...
player 1 complete...
player 4 complete...
player 5 complete...
player 3 complete...
Record the score...

源码分析

The code below relies on pairsAQS的理解,参考：并发编程12-ThreadPoolExecutorUse and principle exploration

初始化方法

// 初始化方法,和其他基于AQS的工具一样,Write an inner classSync继承AQS,Set the value of the counter toAQS的state
public CountDownLatch(int count) {
    
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

Sync(int count) {
    
    setState(count);
}

await()方法

// await()方法,Here, the parent class template method is still called
public void await() throws InterruptedException {
    
    sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    
    if (Thread.interrupted())
        throw new InterruptedException();
    // This defines the template process for acquiring shared locks,The actual logic is left to subclasses to implement,就会走到CountDownLatch的tryAcquireShared(arg)方法中
    // 参考下面的方法,如果tryAcquireShared(arg) > 0,Indicates that the counter has been decremented by another thread0,程序正常执行
    // 如果tryAcquireShared(arg) < 0,Indicates that the counter has not been decremented0,调用await()The thread of the method needs to block itself.
    if (tryAcquireShared(arg) < 0) 
        doAcquireSharedInterruptibly(arg);
}

protected int tryAcquireShared(int acquires) {
    
    // 判断stateThat is, whether the counter is equal to or not0
    return (getState() == 0) ? 1 : -1;
}

countDown()方法

// countDown()方法,Still the same way,调用AQS的releaseShared(1)方法
public void countDown() {
    
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    
    // 老套路,Call the method overridden by the subclass to release the shared lock
    if (tryReleaseShared(arg)) {
    
        doReleaseShared();   // 标记2
        return true;
    }
    return false;
}

protected boolean tryReleaseShared(int releases) {
    
    // Guaranteed by the spin methodCASModify the value operation can be successfully executed
    for (;;) {
    
        int c = getState();
        // 如果已经等于0,直接返回
        if (c == 0)
            return false;
        int nextc = c-1;
        // CAS修改计数器的值,Returns whether the counter is equal0,If get heretrue,Indicates that the current thread decrements the counter to0了,You need to enter the marker2的代码
        // Release is still pendingawait()阻塞的线程
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}