Concurrency tool class CountDownLatch source analysis and usage scenarios

Scan the qr code below or search the wechat official account, cainiao Feiyafei, you can follow the wechat official account, read more Spring source code analysis and Java concurrent programming articles.

Introduction to the

• CountDownLatch is a utility class provided with the JUC package that enables one or a group of threads to wait for other threads to complete their execution. As you can guess from the name, it works by counting backwards and finally opening the latch. That is, after each thread has finished its work, the counter is decreed once. When the counter reaches 0, the lock (latch) is opened and the main thread (or waiting thread) takes over its work.
• In practical work, we may need to use multiple threads to solve problems and control the execution order of these threads. In this case, we can choose to use the join() method provided by Thread class or use CountDownLatch which will be introduced today to solve problems. You can also use CyclicBarrier, another class in the JUC package.

How to use

• The use of CountDownLatch is simple. It has only one constructor. In the constructor, we pass in an int parameter that controls how many times CountDownLatch will decrement to release the latch. CountDownLatch also provides the following three methods, as detailed in the table below.

• Here is a simple scenario to introduce the use of CountDownLatch. In the student’s time, there will always be a variety of tests, each test, each subject teacher will mark, calculate the total score, total score ranking. In this process, each subject is marked at the same time, because the speed of the teachers in each subject is different, so the person who calculates the total score and the total score ranking has to wait until all the teachers have marked the paper. At this point we can use the CountDownLatch utility class to simulate this scenario in our application. The teacher of each subject is treated as a thread. Since the reading speed of the teacher of each subject is different, let the thread sleep randomly for a period of time. When the teacher of each subject finishes reading, call CountDownLatch’s countDown() method to reduce the counter by one. Call the await() method on CountDownLatch so that the main latch waits for all the teachers to finish grading. When all the teachers have finished grading, the counter is reduced to 0, and the main latch unblocks from the await() method, and then sums, ranks, etc. The following is an example.

public class CountDownLatchDemo {

    public static void main(String[] args) {
        CountDownLatch countDownLatch = new CountDownLatch(6);
        List<String> teachers = Arrays.asList("Chinese teacher"."Math teacher"."English teacher"."Physics teacher"."Chemistry Teacher"."Biology teacher");
        Random random = new Random();
        // Create 6 threads to simulate the teachers of 6 subjects at the same time
        List<Thread> threads = new ArrayList<>(6);
        for (int i = 0; i < 6; i++) {
            threads.add(new Thread(()->{
                try {
                    int workTime = random.nextInt(6) + 1;
                    // Let the thread sleep for some time to simulate the teacher's marking time
                    Thread.sleep(workTime * 1000l);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "Marking completed");
                // After each teacher has marked the paper, the counter is reduced by 1
                countDownLatch.countDown();
            },teachers.get(i)));
        }
        for (Thread thread : threads) {
            thread.start();
        }
        // Let the main thread wait for all teachers to finish marking papers before counting the total score and ranking
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("All subjects have been marked by teachers.");
        System.out.println("Start counting points and rank."); }}Copy the code

Source code analysis

• Now that you know how CountDownLatch works, let’s take a look at how CountDownLatch works.
• CountDownLatch is actually a shared lock whose underlying implementation is using the queue synchronizer AQS. CountDownLatch specifies the initial value of the counter in the constructor, the value of the synchronization variable state in AQS. It allows multiple threads to access state at the same time, decreasing the value of state by one each time the countDown() method is called; When the await() method is called, it determines whether the state value is 0, and if it is 0, it tells the current thread to return, and if it is not 0, it tells the current thread to wait in the synchronous queue.
• Since CountDownLatch is implemented using AQS, an internal synchronization component needs to be defined that inherits AQS (this is common practice in the AQS family of locks). An internal class Sync is defined in CountDownLatch that inherits AQS. Methods tryAcquireShared() and tryReleaseShared() in AQS are rewritten.
• When CountDownLatch’s constructor is used to create CountDownLatch, the Sync component is instantiated in the constructor and the value of the Sync variable state in the AQS is initialized through Sync’s parameter constructor. The size of the value is the value of type int passed into the CountDownLatch constructor to indicate the size of the counter.

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}
Copy the code

private static final class Sync extends AbstractQueuedSynchronizer {

    Sync(intcount) { setState(count); }}Copy the code

• When the CountDownLatch’s countDown() method is called, sync.releaseshared (1), the AQS releaseShared() method, is called. The releaseShared() method has the following source code.

public final boolean releaseShared(int arg) {
    // Try to release the shared lock
    if (tryReleaseShared(arg)) {
        /** * When the lock is released, the state of the synchronization is set to 0, indicating that the next thread can acquire the lock. ** * If there is no one waiting, the next thread will wake up. That is, subsequent threads can directly acquire the lock */
        doReleaseShared();
        return true;
    }
    return false;
}
Copy the code

• In the releaseShared() method, the tryReleaseShared() method of the subclass is called first, so the tryReleaseShared() method of Sync, the inner class in CountDownLatch, is called. The source code is as follows.

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    // Subtracting the synchronization status by one, after which the synchronization status becomes 0, returns true, indicating that the waiting thread in the synchronization queue can be awakened
    for (;;) {
        int c = getState();
        If the value of state is 0 and the thread is subtracting state by 1, this is an abnormal operation, so it returns false
        if (c == 0)
            return false;
        int nextc = c-1;
        // Use the CAS operation to set the value of state
        if (compareAndSetState(c, nextc))
            return nextc == 0; }}Copy the code

• The tryReleaseShared() method reduces the state value by 1. If the state value is 0, it returns true, indicating that the waiting thread in the synchronization queue can be awakened, or false if it is not 0. When the tryReleaseShared() method ends, the AQS releaseShared() method is returned. If the tryReleaseShared() method returns false, releaseShared() is returned. If the tryReleaseShared() method returns true, the doReleaseShared() method is then executed. The doReleaseShared() method is a template method defined in AQS that handles the logic of shared locks. Its main purpose is to wake up waiting threads in the synchronization queue.
• When calling the CountDownLatch await (), will call to sync. The acquireSharedInterruptibly (1), namely AQS acquireSharedInterruptibly () method. AcquireSharedInterruptibly source () method are as follows.

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    // The response is interrupted
    if (Thread.interrupted())
        throw new InterruptedException();
    // The tryAcquireShared() method is to try to get the lock
    For CountDownLatch, 1 is returned when state=0, indicating that the lock has been released by all threads, and -1 is returned when state is not 0, indicating that there are still threads holding the lock
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
Copy the code

• In acquireSharedInterruptibly () method will call a subclass tryAcquireShared () method, call here is the inner class CountDownLatch Sync tryAcquireShared () method. The logic of the tryAcquireShared() method is very simple. The tryAcquireShared() method checks whether the state value is 0, if 0, returns 1, if not, returns -1. Said when the state of 0 is the value of the counter is reduced to zero, indicating that all threads have finished their work, so this time tryAcquireShared () method returns 1, then back to AQS, acquireSharedInterruptibly () method directly to the end, The current thread does not block. If state is not 0, the counter has not yet been reduced to 0, and the thread has not yet done its job by calling countDown(), so the tryAcquireShared() method returns -1. AcquireSharedInterruptibly () method does not directly end, but then perform doAcquireSharedInterruptibly () method. DoAcquireSharedInterruptibly () method is a template of AQS method, this method is the main function of dealing with the Shared lock of logic, if the Shared lock acquisition fails, let the thread into the synchronous queue in the park. Here, if the counter is not 0, the current thread calls the await() method and goes to park in the synchronous queue, and will not wake up or be interrupted until another thread calls the countDown() method to reduce the counter to 0.

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

• The implementation logic of the await(long timeout, TimeUnit unit) method is similar to that of the await() method, except that a timeout judgment is added to the await() method. In general, CountDownLatch is relatively simple. It works the same way as shared locks, except that the tryAcquireShared() and tryReleaseShared() methods have slightly different logic.

conclusion

• This article begins with the introduction of CountDownLatch, which allows a thread or group of threads to wait for other threads to complete their execution before running. The purpose of this latch is to control the order in which threads execute. Then this paper demonstrates the use of CountDownLatch through a case of grading, calculating scores and ranking. Finally, the source code of CountDownLatch is simply analyzed. The bottom layer of CountDownLatch is realized according to the related methods of AQS shared lock.
• CountDownLatch is best usedawait(long timeout, TimeUnit unit)Because if the child thread that is processing the task never completes, the countDown() method will never be called, so that the counter does not decrement to zero, causing the main thread to wait and do nothing, it is recommended to use await(long timeout, TimeUnit unit). If in a specific business it is necessary to await the main thread after all threads have finished executing, then use await() method, but in code where child threads are processing tasks, it is better to use try… catch… Finally statement, and then countDown() in the finally block, otherwise it’s easy to dig yourself into a deep hole.
• I once used CountDownLatch and did not use try in the child thread because I used await(). catch… The main thread blocks and waits because a thread processing the business logic failed and did not call the countDown() method. Even worse, since this is an exception in the child thread, there is no try… catch… Finally, this is when the main thread willswallowChild thread stack exception information, resulting in no error log is printed. Since this code is executed when the service is started, the problem is that the service never gets up and the error log is printed. At the time of the problem, the code had been running online for a long time and only appeared in the test environment, so it didn’t even occur to me. In addition to their knowledge of multithreading related mastery of almost 0, check for a long time did not find the reason, the server restarted n times, see the restart method is not good, can only consult colleagues, finally found the error. The final solution is to use a try in the child thread… catch… Finally, and then call countDown() in the finally statement block. In fact, this problem with jstack command, on the server to look at the stack of threads can be found where the problem, but still because of their own very vegetables, do not have enough understanding of multithreading, it took a long time to solve. It was because of this lesson that I decided to start learning concurrency related source code.

recommended

• The implementation principle of ReadWriteLock
• Design principle of queue synchronizer (AQS)
• Queue synchronizer (AQS) source code analysis
• ReentrantLock source code analysis
• Fair locks versus unfair locks
• @ Import and @ EnableXXX