CountDownLatch源码解析以及使用方法

水深无声 2023-02-14 07:53 2阅读 0赞

# CountDownLatch的原理以及使用方法 #

概述

1 ountDownLatch这个类使一个线程等待其他线程各自执行完毕后再执行。

2 是通过一个计数器来实现的，计数器的初始值是线程的数量。每当一个线程执行完毕后，计数器的值就-1，当计数器的值为0时，表示所有线程都执行完毕，然后在闭锁上等待的线程就可以恢复工作了。

源码解析

构造函数

当我们调用CountDownLatch countDownLatch=new CountDownLatch(4) 时候，此时会创建一个AQS的同步队列，并把创建CountDownLatch 传进来的计数器赋值给AQS队列的 state，所以state的值也代表CountDownLatch所剩余的计数次数

//参数count为计数值
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);//创建同步队列，并设置初始计数器值
    }

await()

创建一个节点，加入到AQS阻塞队列，并同时把当前线程挂起。

//调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
    public void await() throws InterruptedException {
    	sync.acquireSharedInterruptibly(1);
    }
    
    //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
    public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    	return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

acquireSharedInterruptibly(int arg)

//1.判断当前线程是否有被中断
    //2.如果没有的话，就调用tryAcquireShared(int acquires)方法，判断当前线程是否还需要“阻塞”。
     public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
    }

doAcquireSharedInterruptibly(int arg)

private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
        	//加入等待队列 				      
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
        	// 进入 CAS 循环
            try {
                for (;;) {
                    //当一个节点(关联一个线程)进入等待队列后， 获取此节点的 prev 节点 
                    final Node p = node.predecessor();
                    // 如果获取到的 prev 是 head，也就是队列中第一个等待线程
                    if (p == head) {
                        // 再次尝试申请 反应到 CountDownLatch 就是查看是否还有线程需要等待(state是否为0)
                        int r = tryAcquireShared(arg);
                        // 如果 r >=0 说明 没有线程需要等待了 state==0
                        if (r >= 0) {
                            //尝试将第一个线程关联的节点设置为 head 
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    //经过自旋tryAcquireShared后，state还不为0，就会到这里，第一次的时候，waitStatus是0，那么node的waitStatus就会被置为SIGNAL，第二次再走到这里，就会用LockSupport的park方法把当前线程阻塞住
                    //重组双向链表，清空无效节点，挂起当前线程
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
     }

总结

1 当前线程就会进入了一个死循环当中，在这个死循环里面，会不断的进行判断

2 通过调用tryAcquireShared方法，不断判断我们上面说的那个计数器，

3 看看它的值是否为0了（为0的时候，其实就是我们调用了足够多次数的countDownLatch.countDown（）方法的时候）

4 如果是为0的话，tryAcquireShared就会返回1，设置第一个线程关联的借点未head，然后跳出了循环，不再“阻塞”当前线程了。

5 需要注意的是，说是在不停的循环，其实也并非在不停的执行for循环里面的内容，因为在后面调用parkAndCheckInterrupt（）方法时，在这个方法里面是会调用 LockSupport.park(this)来禁用当前线程的。

parkAndCheckInterrupt()

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

addWaiter(Node mode)

将当前线程加入等待队列

static final Node SHARED = new Node();
    
    static final Node EXCLUSIVE = null;
    
    private Node addWaiter(Node mode) {
            Node node = new Node(Thread.currentThread(), mode);
            // 尝试快速入队操作，因为大多数时候尾节点不为 null
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
        	//如果尾节点为空(也就是队列为空) 或者尝试CAS入队失败(由于并发原因)，进入enq方法
            enq(node);
            return node;
    }

步骤：

1.构造Node实体，参数为当前线程和mode，mode是SHARED 或者 EXCLUSIVE

2.尝试快速入队列操作

3.如果尾节点为空(也就是队列为空) 或者尝试CAS入队失败(由于并发原因)，进入enq方法

enq(final Node node)

private Node enq(final Node node) {
        	// 死循环+CAS保证所有节点都入队
            for (;;) {
                Node t = tail;
                // 如果队列为空 设置一个空节点作为 head
                if (t == null) { // Must initialize
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    //加入队尾
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
     }

说明：

1.死循环+CAS 乐观锁

2.CAS是实现原子性，compareAndSetTail底层调用的是unsafe类，直接操作内存，在cpu层上加锁，直接对内存进行操作

setHeadAndPropagate(node, r)

private void setHeadAndPropagate(Node node, int propagate) {
        //备份head
        Node h = head; 
        //抢到锁的线程被唤醒 将这个节点设置为head
        setHead(node);
        // propagate 一般都会大于0 或者存在可被唤醒的线程
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
             // 只有一个节点 或者是共享模式 释放所有等待线程 各自尝试抢占锁
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

Node 中的waitStatus四种状态

//线程已被 cancelled ，这种状态的节点将会被忽略，并移出队列
    static final int CANCELLED =  1;
    // 表示当前线程已被挂起，并且后继节点可以尝试抢占锁
    static final int SIGNAL    = -1;
    //线程正在等待某些条件
    static final int CONDITION = -2;
    //共享模式下 无条件所有等待线程尝试抢占锁
    static final int PROPAGATE = -3;

countDown()

//将计数值减一
    //递减锁重入次数，当state=0时唤醒所有阻塞线程
    public void countDown() {
            sync.releaseShared(1);
    }

releaseShared(1)

// AQS类
    public final boolean releaseShared(int arg) {
        	// arg 为固定值 1
        	// 如果计数器state 为0 返回true，前提是调用 countDown() 之前不能已经为0
            if (tryReleaseShared(arg)) {
                // 唤醒等待队列的线程
                doReleaseShared();
                return true;
            }
            return false;
    }
    
    // CountDownLatch 重写的方法
    protected boolean tryReleaseShared(int releases) {
                // Decrement count; signal when transition to zero
        		// 依然是循环+CAS配合 实现计数器减1
                for (;;) {
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
    }
    
    /// AQS类
     private void doReleaseShared() {
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    // 如果节点状态为SIGNAL，则他的next节点也可以尝试被唤醒
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);//成功则唤醒线程
                    }
                    // 将节点状态设置为PROPAGATE，表示要向下传播，依次唤醒
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
    }

## **总结** ##

1、AQS 分为独占模式和共享模式，CountDownLatch 使用了它的共享模式。

2、AQS 当第一个等待线程（被包装为 Node）要入队的时候，要保证存在一个 head 节点，这个 head 节点不关联线程，也就是一个虚节点。

3、当队列中的等待节点（关联线程的，非 head 节点）抢到锁，将这个节点设置为 head 节点。

4、第一次自旋抢锁失败后，waitStatus 会被设置为 -1（SIGNAL），第二次再失败，就会被 LockSupport 阻塞挂起。

5、如果一个节点的前置节点为 SIGNAL 状态，则这个节点可以尝试抢占锁。

## **实现逻辑** ##

1、初始化CountDownLatch实际就是设置了AQS的state为计数的值

2、调用CountDownLatch的countDown方法时实际就是调用AQS的释放同步状态的方法，每调用一次就自减一次state值

3、调用await方法实际就调用AQS的共享式获取同步状态的方法acquireSharedInterruptibly(1)，这个方法的实现逻辑就调用子类Sync的tryAcquireShared方法，只有当子类Sync的tryAcquireShared方法返回大于0的值时才算获取同步状态成功，否则就会一直在死循环中不断重试，直到tryAcquireShared方法返回大于等于0的值，而Sync的tryAcquireShared方法只有当AQS中的state值为0时才会返回1，否则都返回-1，也就相当于只有当AQS的state值为0时，await方法才会执行成功，否则就会一直处于死循环中不断重试。

示例

public class CountDownLatchDemo {
        public static void main(String[] args) throws InterruptedException {
            /*创建CountDownLatch实例,计数器的值初始化为5*/
            final CountDownLatch downLatch = new CountDownLatch(5);
    
            /*每个线程等待1s,表示执行比较耗时的任务*/
            for(int i = 0;i < 5;i++){
                final int num = i;
                new Thread(new Runnable() {
                    public void run() {
                        try {
                            Thread.sleep(1000);
    
                        }catch (InterruptedException e){
                            e.printStackTrace();
    
                        }
    
                        System.out.println(String.format("thread %d has finished",num));
    
                        /*任务完成后调用CountDownLatch的countDown()方法*/
                        downLatch.countDown();
                    }
                }).start();
            }
    
            /*主线程调用await()方法,等到其他5个线程执行完后才继续执行*/
            downLatch.await();
    
            System.out.println("all threads have finished,main thread will continue run");
        }
    }

输出

thread 1 has finished
    thread 2 has finished
    thread 0 has finished
    thread 3 has finished
    thread 4 has finished
    all threads have finished,main thread will continue run

**使用场景**

1 需要等待某个条件达到要求后才能做后面的事情

2 同时当线程都完成后也会触发事件，以便进行后面的操作

**如果大家对java架构相关感兴趣，可以关注下面公众号，会持续更新java基础面试题, netty, spring boot,spring cloud等系列文章，一系列干货随时送达, 超神之路从此展开, BTAJ不再是梦想!**

![架构殿堂][20200415234301759.jpg]

[20200415234301759.jpg]: https://img-blog.csdnimg.cn/20200415234301759.jpg