1什么是Latch

若干线程并发执行某个特定的任务，然后等到所有的子任务都执行结束之后再统一汇总，比如用户想要查询自己三年以来银行账号的流水，为了保证运行数据库的数据量在一个恒定的范围之内，通常数据只会保存一年的记录，其他的历史记录或被备份到磁盘，或者被存储于hive数据仓库，或者被转存至备份数据库之中，总之想要三年的流水记录，需要若干个渠道的查询才可以汇齐。如果一个线程负责执行这样的任务，则需要经历若干次的查询最后汇总返回给用户，很明显这样的操作性能低下，用户体验差，如果我们将每一个渠道的查询交给一个线程或者若干个线程去查询，然后统一汇总，那么性能会提高很多，响应时间也会缩短不少。

Latch（门阀）设计模式，该模式指定了一个屏障，只有所有的条件都达到满足的时候，门阀才能打开。

2 CountDownLatch实现

2.1 无限等待的Latch

1. 首先定义了一个无限等待的抽象类Latch，在Latch抽象类中定义了await方法、countDown方法以及getUnarrived方法

public abstract class Latch {
    /**
     * 用于控制多少个线程完成任务时，才能打开阀门
     */
    protected int limit;

    public Latch(int limit) {
        this.limit = limit;
    }

    /**
     * 抽象方法，该方法会使当前线程一直等待，直到所有
     * 线程都完成工作，被阻塞的线程的是允许被中断的
     */
    public abstract void await() throws InterruptedException;

    /**
     * 当任务线程完成之后调用该方法使计数器减一
     */
    public abstract void countDown();

    /**
     * 获取当前还有多少个线程没有完成任务
     * @return
     */
    public abstract int getUnArrived();
}

实现这些抽象方法

public class CountDownLatch extends Latch {

    public CountDownLatch(int limit) {
        super(limit);
    }

    @Override
    public void await() throws InterruptedException {
        synchronized (this) {
            // 当 limit > 0的时候，当前线程进入阻塞状态
            while (this.limit > 0) {
                this.wait();
            }
        }
    }

    @Override
    public void countDown() {
        synchronized (this) {
            if (limit <= 0) {
                throw new IllegalArgumentException("all of task already arrived");
            }
            // limit -1, 并通知唤醒阻塞线程
            this.limit--;
            this.notifyAll();
        }
    }

    @Override
    public int getUnArrived() {
        // 返回当前还有多少个线程没有完成任务
        return this.limit;
    }
}

测试

public static void main(String[] args) throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(3);
    // 定义3个线程去，只有当三个线程都执行完了，主线程才会继续
    Thread thread1 = new Thread("线程1") {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " start");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " end");
            countDownLatch.countDown();
        }
    };
    Thread thread2 = new Thread("线程2") {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " start");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " end");
            countDownLatch.countDown();
        }
    };
    Thread thread3 = new Thread("线程3") {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + " start");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " end");
            countDownLatch.countDown();
        }
    };
    thread1.start();
    thread2.start();
    thread3.start();
    // 等待三个线程执行结束
    countDownLatch.await();
    System.out.println("三个线程已经全部执行完毕");
}

线程2 start
线程1 start
线程3 start
线程1 end
线程2 end
线程3 end
三个线程已经全部执行完毕

2.2 增加超时设置

可以设置超时时间，最多等待多长时间，就不等了。

在抽象类中增加如下方法：

/**
 * 支持超时
 */
public abstract void await(TimeUnit timeUnit,long time) throws InterruptedException, TimeoutException;

实现：

@Override
public void await(TimeUnit timeUnit, long time) throws InterruptedException, TimeoutException {
    if (time <= 0) {
        throw new IllegalArgumentException("The time is invalid.");
    }
    // 剩余等待时间：将时间统一转化为纳秒
    long remainingNanos = timeUnit.toNanos(time);
    // 计算超时结束时间
    long endNanos = System.nanoTime() + remainingNanos;
    synchronized (this) {
        while (limit > 0) {
            // 如果超时了(剩余等待时间小于0)，就抛出超时异常
            if (TimeUnit.NANOSECONDS.toMillis(remainingNanos) <= 0) {
              throw new TimeoutException("等待超时");
            }
            // 如果没有超时，那么就等待remainingNanos
            this.wait(TimeUnit.NANOSECONDS.toMillis(remainingNanos));
            // 在等待的过程中有可能会被中断，需要重新计算remainingNanos: 超时结束时间 - 减去当前时间
            remainingNanos = endNanos - System.nanoTime();
        }
    }
}

测试：将之前测试换成超时等待的方法

// 等待三个线程执行结束
try {
    countDownLatch.await(TimeUnit.SECONDS,2);
} catch (TimeoutException e) {
    System.out.println("超时了，不在等待了，主线程继续执行");
}

线程2 start
线程3 start
线程1 start
线程1 end
线程2 end
超时了，不在等待了，主线程继续执行
三个线程已经全部执行完毕
线程3 end

3 CountDownLatch完整代码

public class CountDownLatch extends Latch {

    public CountDownLatch(int limit) {
        super(limit);
    }

    @Override
    public void await() throws InterruptedException {
        synchronized (this) {
            // 当 limit > 0的时候，当前线程进入阻塞状态
            while (this.limit > 0) {
                this.wait();
            }
        }
    }

    @Override
    public void await(TimeUnit timeUnit, long time) throws InterruptedException, TimeoutException {
        if (time <= 0) {
            throw new IllegalArgumentException("The time is invalid.");
        }
        // 剩余等待时间：将时间统一转化为纳秒
        long remainingNanos = timeUnit.toNanos(time);
        // 计算超时结束时间
        long endNanos = System.nanoTime() + remainingNanos;
        synchronized (this) {
            while (limit > 0) {
                // 如果超时了(剩余等待时间小于0)，就抛出超时异常
                if (TimeUnit.NANOSECONDS.toMillis(remainingNanos) <= 0) {
                  throw new TimeoutException("等待超时");
                }
                // 如果没有超时，那么就等待remainingNanos
                this.wait(TimeUnit.NANOSECONDS.toMillis(remainingNanos));
                // 在等待的过程中有可能会被中断，需要重新计算remainingNanos: 超时结束时间 - 减去当前时间
                remainingNanos = endNanos - System.nanoTime();
            }
        }
    }

    @Override
    public void countDown() {
        synchronized (this) {
            if (limit <= 0) {
                throw new IllegalArgumentException("all of task already arrived");
            }
            // limit -1, 并通知唤醒阻塞线程
            this.limit--;
            this.notifyAll();
        }
    }

    @Override
    public int getUnArrived() {
        // 返回当前还有多少个线程没有完成任务
        return this.limit;
    }
}

介绍

CountDownLatch是一个同步辅助类，它允许一个或多个线程一直等待直到其他线程执行完毕才开始执行。

用给定的计数初始化CountDownLatch，其含义是要被等待执行完的线程个数。

每次调用CountDown()，计数减1

主程序执行到await()函数会阻塞等待线程的执行，直到计数为0

场景

1:多线程读取批量文件, 并且读取完成之后汇总处理

2:多线程读取Excel多个sheet,读取完成之后获取汇总获取的结果

3:多个人一起一起来吃饭,主人等待客人到来,客人一个个从不同地方来到饭店,主人需要等到所有人都到来之后,才能开饭

4:汽车站,所有乘客都从不同的地方赶到汽车站,必须等到所有乘客都到了,汽车才会出发,(如果设置了超时等待,那么当某个时间点到了,汽车也出发)

5: 百米赛跑，4名运动员选手到达场地等待裁判口令，裁判一声口令，选手听到后同时起跑，当所有选手到达终点，裁判进行汇总排名

6: 4名选手进行大众投票,投票现场有500个票, 投票时间是30分钟,500个票投完结束投票,或者投票时间到了也结束投票

作用：可以用来确保某些活动直到其他活动都完成后才继续执行。

CountDownLatch非常适合于对任务进行拆分，使其并行执行，比如某个任务执行2s，其对数据的请求可以分为五个部分，那么就可以将这个任务拆分为5个子任务，分别交由五个线程执行，执行完成之后再由主线程进行汇总，此时，总的执行时间将决定于执行最慢的任务，平均来看，还是大大减少了总的执行时间。

注意事项:

1. 使用CountDownLatch必须确保计数器数量与子线程数量一致,且countDown必须要执行,否则出现计数器不为0,导致主线程一致等待的情况

2. 在执行任务的线程中，使用了try...finally结构，该结构可以保证创建的线程发生异常时CountDownLatch.countDown()方法也会执行，也就保证了主线程不会一直处于等待状态。

案例(模拟乘客登机的场景)

    public static void main(String[] args) {
        CountDownLatchUtils.initialize("da1", 5);
        List<String> list=new CopyOnWriteArrayList<>();

        ExecutorUtils.create(()->{
            System.out.println("张三正在马鞍山,准备赶到南京坐飞机,需要1小时的车程到机场");
            SleepTools.second(4);
            list.add("张三");
            CountDownLatchUtils.countDown("da1");
        });
        ExecutorUtils.create(()->{
            System.out.println("李四正在徐州,准备赶到南京坐飞机,需要5小时的车程到机场");
            SleepTools.second(15);
            list.add("李四");
            CountDownLatchUtils.countDown("da1");
        });

        ExecutorUtils.create(()->{
            System.out.println("王五正在芜湖,准备赶到南京坐飞机,需要2小时的车程到机场");
            SleepTools.second(9);
            list.add("王五");
            CountDownLatchUtils.countDown("da1");
        });

        ExecutorUtils.create(()->{
            //飞机起飞
            CountDownLatchUtils.await("da1",10);  //这里先模拟10秒 360秒=1小时
            System.out.println("南京禄口机场_机长启动飞机起飞");
            //当前航班已到乘客
            System.out.println("当前航班已到乘客"+list);
        });
    }

以上代码都是我封装后的,主要是看逻辑就行了

张三正在马鞍山,准备赶到南京坐飞机,需要1小时的车程到机场
李四正在徐州,准备赶到南京坐飞机,需要5小时的车程到机场
王五正在芜湖,准备赶到南京坐飞机,需要2小时的车程到机场
南京禄口机场_机长启动飞机起飞
当前航班已到乘客[张三, 王五]