Countdownlatch source code analysis

One 、CountDownLatch brief introduction

CountDownLatch( atresia ) Is a synchronization assistance class , Allow one or more threads to block waiting , Until the other threads have finished , The blocked thread will be awakened , Then execute the following business logic .

CountDownLatch The construction method of needs to give a count value (count) To initialize .CountDownLatch A simple understanding is a counter , Call it the await() Will be blocked , Until its count value (count) by 0 when , The blocked thread will be awakened , And call it countDown() Method will set the counter value -1, Until the count value is 0 after , Previously blocked threads will be awakened , At the same time, the following calls will be made await() Method will no longer be blocked , Because the count count It's disposable , When its value is reduced to 0 It won't change after , So subsequent calls await() Method will not be blocked , It's about going back to .

If you are using , It is required that the count value can be reset , Consider using CyclicBarrier, About CyclicBarrier It will be introduced in detail in the next article .

The following figure shows CountDownLatch Principle ：

When count Values for 0 The thread will be restored A Carry on

Two 、 Use scenarios

CountDownLatch Generally used for multi-threaded countdown counters , Force them to wait for another set of threads ( from CountDownLatch Initialization decision ) Task execution completed .

CountDownLatch There are two usage scenarios ：

• Let multiple threads wait
• Let a single thread wait

2.1 Multithread waiting ： Concurrent threads execute simultaneously

This kind of scene is very similar to the track and field mobilization competition , Every athlete is a thread , Then the caller is a main thread , After each remote mobilization is ready, you need to wait for the order from the dispatcher to start , The dispatcher fired when he was ready , Then all the far mobilization starts running at the same time .

The following code simulates the scene of the far mobilization game ：

Three remote mobiles are blocked in the call await() Method here , When the dispatcher is ready to give the order in two seconds , These are three far mobilization going forward at the same time

CountDownLatch countDownLatch = new CountDownLatch(1);

for (int num = 0; num < 3; num++){
    
    new Thread(() ->{
    
        try {
    
            // Ready , The athletes are stuck here , Waiting for orders 
            countDownLatch.await();
            System.out.println(Thread.currentThread().getName()+" Start running ……");
        } catch (InterruptedException e) {
    
            log.error(e);
        }
    }).start();
}
//  The dispatcher is ready to give the order 
Thread.sleep(2000);
//  starting gun ： A starting 
countDownLatch.countDown();

2.2 Single thread wait ： Merge after concurrent threads are completed

In some scenes , There are dependencies before and after concurrent tasks , For example, the data details page needs to call multiple interfaces to obtain data at the same time , After the concurrent request obtains the data , The results need to be merged . This is a typical scenario where concurrent threads need to be merged after completion .

Wait for the merged thread to call await() Method to block and wait , When each task is completed, it will call countDown() Method , Will count -1, When the last thread finishes executing the task , The count value is reduced to 0, At this time, wake up the thread waiting for the summary .

CountDownLatch countDownLatch = new CountDownLatch(3);
for (int num = 0; num < 3; num++){
    
    new Thread(() ->{
    
        try {
    
            Thread.sleep(1000+ ThreadLocalRandom.current().nextInt(1000));
            System.out.println(Thread.currentThread().getName() + "finish task");

            countDownLatch.countDown();
        } catch (InterruptedException e) {
    
            log.error(e);
        }
    }).start();
}

countDownLatch.await();
System.out.println(" All threads complete , Summarize the results ");

3、 ... and 、 Source code analysis

Understanding CountDownLatch Source code before , You can have a look at 《Semaphore Source code analysis 》 This article , I understand later CountDownLatch The source code is very simple .

CountDownLatch Compared with the source code of ReentrantLock and Semaphore It's a lot easier , Because it is a function to the counter , Therefore, there will be no concept of fairness and unfairness , At the same time, its underlying implementation is similar to Semaphore It's exactly the same , It just rewrites tryAcquireShared() and tryReleaseShared() Two methods

CountDownLatch Also based on AbstractQueuedSynchronizer Realized , So the inner class Sync Inherited AbstractQueuedSynchronizer, And then use AQS Of state Property to record count The count of .

CountDownLatch The whole structure of is very simple , It only provides the following methods ：

//  The constructor needs to specify count value 
public CountDownLatch(int count) {
    
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

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

//  Block waiting timeout After a long time ,count Value hasn't changed to 0, No more waiting , Carry on 
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

//  take count value -1, Until for 0
public void countDown() {
    
    sync.releaseShared(1);
}

public long getCount() {
    
    return sync.getCount();
}

3.1 Block waiting

In the source code of the above method , You can see await() Method is the inner class of the call Sync Of acquireSharedInterruptibly() Method , The method in AQS Has been implemented in ,Semaphore This method is also called when obtaining shared resources

public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

The core of blocking is tryAcquireShared(), As long as the return value of this method is less than -1, Will call doAcquireSharedInterruptibly() Method blocks the thread

and CountDownLatch The inner class of Sync Chinese vs tryAcquireShared() The implementation of the method is also very simple , as long as count The value of is not equal to 0, Just go back to -1, Indicates that blocking is required .

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

Then you call doAcquireSharedInterruptibly() Blocking threads , The method is described in detail in Semaphore The article on source code analysis has been introduced in detail .

3.2 Wake up to perform

CountDownLatch Intermediate cause call awati() Method blocked thread , Whether you can be awakened depends entirely on countDown() Method , This method will make count The count of -1.

CountDownLatch Of countDown() Method is called Sync Of releaseShared() Method , Again , This method is also used in AQS Has been realized in

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

Can I call doReleaseShared() Wake up the thread , Depending on CountDownLatch in Sync Of tryReleaseShared() Method

In this method , First judge count Is the value of 0, If 0 It means that you have been awakened , There is no need to wake up repeatedly , So go straight back false

If not for 0, Just use spin +CAS take state The attribute value -1, The modified count The value is 0, It will return true

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

return true after , Will call AQS Of doReleaseShared() Wake up the thread , This method has been used in Semaphore The source code of .

From above CountDownLatch The source code of , It's all about Semaphore Characteristics of ( The awakened thread will try to wake up the subsequent thread ) To wake up all blocked threads

summary ： In terms of function ,CountDownLatch With the function of Thread.join() The method is very similar , It is to wait for other threads to complete execution before executing subsequent logic , But in practice ,CountDownLatch Provides a ratio join() The method is more flexible API. meanwhile CountDownLatch That is, you can manually control calling multiple times in a thread countDown() Method , You can also call multiple times in multiple threads .join() The implementation principle of this method is to keep checking join Whether the thread exists , Keep the current thread blocked as long as it is alive .