并发编程（八）：AQS之CountDownLatch

今天药忘吃喽~ 2023-06-04 10:59 20阅读 0赞

# 一，底层AQS源码分析：[并发编程（四）：AbstractQueuedSynchronizer源码分析][AbstractQueuedSynchronizer] #

# 二，CountDownLatch介绍 #

##     1，线程控制 ##

\* *CountDownLatch* 是一个同步工具类，允许一个或者多个线程一直等待，直到其他线程的操作执行完毕再执行。从命令可以解读 *CountDown* 是倒数的意思，类似于倒计时的概念。在API层面，*CountDownLatch *在初始化时需要一个 *count *参数作为初始计数值，线程调用 *await() *方法会阻塞等待，直到存在其他线程调用 *countDown()* 方法，将 *count *值减为0，所有被 *CountDownLatch* 阻塞的线程才会被全部唤醒，此处也是基于共享锁概念。

##     2，类图 ##

![2019100922303132.png][]

\* *CountDownLatch* 内部定义内部类 *Sync*，该类继承自AQS，对AQS部分功能进行扩展。

##     3，常用API ##

// CountDownLatch初始化，传递count参数
    public CountDownLatch(int count);
    // 线程等待
    public void await() throws InterruptedException;
    // 线程定时等待
    public boolean await(long timeout, TimeUnit unit);
    // 计数器递减
    public void countDown();

##     4，功能DEMO ##

package com.gupao;
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * @author pj_zhang
     * @create 2019-09-15 11:33
     */
    public class CountDownLatchTest {
    
        public static void main(String[] args) throws InterruptedException {
            // 初始化CountDownLatch，并传递count为3
            CountDownLatch countDownLatch = new CountDownLatch(3);
            // 定义三个线程，逐步递减，保证countDownLatch能够减为0并释放锁
            for (int i = 0; i < 3; i++) {
                new Thread(() -> {
                    System.out.println(Thread.currentThread().getName() + "正在执行");
                    try {
                        // 每一个线程内部等待
                        // 在 countDownLatch 减为0前，线程会一直在此等待
                        // 等 countDownLatch 减为0后，会释放共享锁，所有被阻塞线程执行
                        countDownLatch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + "执行完毕");
                }, "thread_" + i).start();
                Thread.sleep(100);
                // 每一次循环结束，线程递减
                countDownLatch.countDown();
            }
        }
    }

![20191009224054956.png][]

# 三，CountDownLatch源码分析 #

##     1，*CountDownLatch* 初始化 ##

##     2，等待源码分析 ##

###         2.1，不带时间等待 ###

\* await()：等待

public void await() throws InterruptedException {
    	// 此处调用AQS的尝试获取共享锁代码
    	sync.acquireSharedInterruptibly(1);
    }

\* acquireSharedInterruptibly(int arg)：尝试获取共享锁

public final void acquireSharedInterruptibly(int arg)
    		throws InterruptedException {
    	// 判断线程是否已经被中断
    	if (Thread.interrupted())
    		throw new InterruptedException();
    	// 尝试获取共享锁，此处CountDownLatch自定义实现
    	if (tryAcquireShared(arg) < 0)
    		// 获取锁失败，则进行AQS同步队列处理
    		doAcquireSharedInterruptibly(arg);
    }

\* tryAcquireShared(int acquires)：尝试获取共享锁，该部分 *CountDownLatch* 自定义实现，根据 *state* 状态进行判断

protected int tryAcquireShared(int acquires) {
    	// 获取state状态，并判断是否等于0
    	// 等于0，返回1 > 0，则获取锁成功，继续执行后续代码
    	// 不等于0，返回-1 < 0，则获取锁失败，进行AQS队列处理
    	return (getState() == 0) ? 1 : -1;
    }

\* doAcquireSharedInterruptibly(int arg)：该部分功能已经在AQS源码解析中分析，代码基本一致

###         2.2，带时间等待 ###

\* await(long timeout, TimeUnit unit)：限时等待

public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
    	// 调用AQS的超时获取锁处理
    	return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

\* tryAcquireSharedNanos(int arg, long nanosTimeout)：限时获取锁

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
    		throws InterruptedException {
    	// 判断线程是否中断
    	if (Thread.interrupted())
    		throw new InterruptedException();
    	// 尝试获取共享锁，成功则继续执行，失败执行||后代码，进行AQS同步队列处理
    	return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout);
    }

\* *tryAcquireShared(int acquires)* 和 *doAcquireSharedNanos(int arg, long nanosTimeout) *已经分别分析

##     3，递减直到唤醒源码分析 ##

\* countDown()：通过 *count* 递减，进行锁释放

public void countDown() {
    	// 通过AQS进行共享锁释放，每次重入次数释放为1
    	sync.releaseShared(1);
    }

\* releaseShared(int arg)：释放共享锁，对重入次数进行递减，此处arg恒为1

public final boolean releaseShared(int arg) {
    	// 尝试释放锁，该方法CountDownLatch重新实现
    	if (tryReleaseShared(arg)) {
    		// 进行锁释放，该部分纯AQS实现，之前已经分析
    		doReleaseShared();
    		return true;
    	}
    	return false;
    }

\* tryReleaseShared(int releases)：尝试释放共享锁

protected boolean tryReleaseShared(int releases) {
    	for (;;) {
    		// 获取当前的state，即count，count如果已经为0，则说明已经释放，不进行多次释放
    		int c = getState();
    		if (c == 0)
    			return false;
    		// 此处按道理应该减releases，并通过CAS进行值替换
    		int nextc = c-1;
    		if (compareAndSetState(c, nextc))
    			// 返回判断值是否为0，如果为0，则返回true，进行锁释放
    			return nextc == 0;
    	}
    }

\* doReleaseShared()：共享锁链式释放，此处AQS源码内部已经分析

[AbstractQueuedSynchronizer]: https://blog.csdn.net/u011976388/article/details/101717658
[2019100922303132.png]: /images/20230601/7e26f1cc84ec4cebafd1373a89cb378f.png
[20191009224054956.png]: /images/20230601/59729c5458594322b3f52dee4472a91a.png