014-线程同步辅助类-CountDownLatch

Posted 2021-02-08 bjlhx

一、概述　　

　　CountDownLatch是JAVA提供在java.util.concurrent包下的一个辅助类，指定的一个或多个线程等待其他线程执行完成后执行。

　　能够使一个线程等待其他线程完成各自的工作后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

1.1、主要方法

// 构造器，必须指定一个大于零的计数
public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

// 线程阻塞，直到计数为0的时候唤醒；可以响应线程中断退出阻塞
public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

// 线程阻塞一段时间，如果计数依然不是0，则返回false；否则返回true
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

// 计数-1
public void countDown() {
    sync.releaseShared(1);
}

// 获取计数
public long getCount() {
    return sync.getCount();
}

　　　　它的内部有一个辅助的内部类：sync

　　countDownLath无法被重置，循环栅栏可以。

　　循环栅栏是要执行的线程增加，countDownLatch是执行完成的线程增加。

1.2、实现原理

　　同ReentrantLock一样，依然是借助AQS的双端队列，来实现原子的计数-1，线程阻塞和唤醒

　　参看：013-AbstractQueuedSynchronizer-用于构建锁和同步容器的框架

1. 计数器的初始化

　　CountDownLatch内部实现了AQS，并覆盖了tryAcquireShared()和tryReleaseShared()两个方法，下面说明干嘛用的

　　通过前面的使用，清楚了计数器的构造必须指定计数值,这个直接初始化了 AQS内部的state变量

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

　　后续的计数-1/判断是否可用都是基于state进行的

2. countDown() 计数-1的实现

// 计数-1
public void countDown() {
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) { // 首先尝试释放锁
        doReleaseShared();
        return true;
    }
    return false;
}

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0) //如果计数已经为0，则返回失败
            return false;
        int nextc = c-1;
        // 原子操作实现计数-1
        if (compareAndSetState(c, nextc)) 
            return nextc == 0;
    }
}

// 唤醒被阻塞的线程
private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) { // 队列非空，表示有线程被阻塞
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) { 
            // 头结点如果为SIGNAL,则唤醒头结点下个节点上关联的线程，并出队
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head) // 没有线程被阻塞，直接跳出
            break;
    }
}

countDown内部实现流程:

1. 尝试释放锁tryReleaseShared，实现计数-1

• 若计数已经小于0，则直接返回false
• 否则执行计数(AQS的state)减一
• 若减完之后，state==0，表示没有线程占用锁，即释放成功，然后就需要唤醒被阻塞的线程了

2. 释放并唤醒阻塞线程 doReleaseShared

• 如果队列为空，即表示没有线程被阻塞（也就是说没有线程调用了 CountDownLatch#wait()方法），直接退出
• 头结点如果为SIGNAL, 则依次唤醒头结点下个节点上关联的线程，并出队

注意：即CountDownLatch计数为0之后，所有被阻塞的线程都会被唤醒，且彼此相对独立，不会出现独占锁阻塞的问题

3. await() 阻塞等待计数为0

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

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted()) // 若线程中端，直接抛异常
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}


// 计数为0时，表示获取锁成功
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

// 阻塞，并入队
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED); // 入队
    boolean failed = true;
    try {
        for (;;) {
            // 获取前驱节点
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 获取锁成功，设置队列头为node节点
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) // 线程挂起
              && parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

阻塞的内部实现逻辑

1. 判断state计数是否为0，不是，则直接放过执行后面的代码
2. 大于0，则表示需要阻塞等待计数为0
3. 当前线程封装Node对象，进入阻塞队列
4. 然后就是循环尝试获取锁，直到成功（即state为0）后出队，继续执行线程后续代码

1.3、使用注意

• 在创建实例时，必须指定初始的计数值，且应大于0
• 必须有线程中显示的调用了countDown()计数-1方法；必须有线程显示调用了await()方法（没有这个就没有必要使用CountDownLatch了）
• 由于await()方法会阻塞到计数为0，如果在代码逻辑中某个线程漏掉了计数-1，导致最终计数一直大于0，直接导致死锁了；
• 鉴于上面一点，更多的推荐 await(long, TimeUnit)来替代直接使用await()方法，至少不会造成阻塞死只能重启的情况
• 允许多个线程调用await方法，当计数为0后，所有被阻塞的线程都会被唤醒

1.4、使用场景

• 确保某个计算在其需要的所有资源都被初始化之后才继续执行。
• 确保某个服务在其依赖的所有其他服务都已启动后才启动。
• 等待知道某个操作的所有者都就绪在继续执行。

1.5、示例

　　有5个运动员赛跑，为了公平所有运动员需要同时起跑。同时，当最后一个运动员跑完之后，比赛结束。 

    public static void main(String[] args) {
        //所有线程阻塞，然后统一开始
        CountDownLatch begin = new CountDownLatch(1);
        //主线程阻塞，直到所有分线程执行完毕
        CountDownLatch end = new CountDownLatch(5);
        for(int i = 0; i < 5; i++){
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        begin.await();
                        System.out.println(LocalDateTime.now()+":"+Thread.currentThread().getName() + " 起跑");
                        Thread.sleep(new Random().nextInt(2000));
                        System.out.println(LocalDateTime.now()+":"+Thread.currentThread().getName() + " 到达终点");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    finally {
                        end.countDown();
                    }

                }
            });

            thread.start();
        }

        try {
            System.out.println(LocalDateTime.now()+":"+"1秒后统一开始");
            Thread.sleep(1000);
            begin.countDown();

            end.await();
            System.out.println(LocalDateTime.now()+":"+"停止比赛");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

结果

2019-02-12T16:51:38.903:1秒后统一开始
2019-02-12T16:51:39.904:Thread-1 起跑
2019-02-12T16:51:39.904:Thread-4 起跑
2019-02-12T16:51:39.904:Thread-3 起跑
2019-02-12T16:51:39.904:Thread-2 起跑
2019-02-12T16:51:39.904:Thread-0 起跑
2019-02-12T16:51:40.035:Thread-0 到达终点
2019-02-12T16:51:40.484:Thread-2 到达终点
2019-02-12T16:51:41.087:Thread-4 到达终点
2019-02-12T16:51:41.215:Thread-3 到达终点
2019-02-12T16:51:41.287:Thread-1 到达终点
2019-02-12T16:51:41.288:停止比赛