博客

1.介绍

CountDownLatch类位于java.util.concurrent包下，利用它可以实现类似计数器的功能。比如我们发送多个请求，而其中有一个任务需要多个请求完成之后才会执行，这个时候可以选择使用CountDownLatch计数器功能。它还可以用来提高效率，下面会介绍。

2.源码分析

//计数器减一
public void countDown() {
    sync.releaseShared(1);
}
//调用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));
}

上面三个方法是最常用的。

3.案例

假设有一个这样的场景：目前有三个请求，包装成一个List，批次处理请求，假设每一个请求花1秒钟，耗时的方法。

未使用CountDownLatch

package com.example.demo.thread;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;

import java.util.List;

/**
 * @Description: 未使用CountDown处理多个请求
 * @Author: dingguo
 * @Date: 2019/6/17 晚上8:09
 */
public class Worker {

    private static final Logger logger = LoggerFactory.getLogger(Worker.class);

    /**
     * 模拟批次请求接口
     *
     * @param request
     */
    public List<String> dealTask(List<String> request) {
        StopWatch watch = new StopWatch();
        watch.start();

        List<String> result = Lists.newArrayList();
        request.forEach(s -> {
            try {
                //假设每一个请求都花了1s
                Thread.sleep(1000);
                result.add("返回的结果:" + s);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        watch.stop();
        logger.info("花费时间:[{}]s", watch.getTotalTimeMillis());
        return result;
    }

    public static void main(String[] args) {
        Worker worker = new Worker();
        List<String> list = Lists.newArrayList("第一次请求", "第二次请求", "第三次请求");
        List<String> result = worker.dealTask(list);
        //会使用批量获取的结果值
        logger.info("处理批量返回的数据：{}", result);
    }
}

执行之后发现总花费时间是3秒钟，并没有高效率。

将代码进行优化，使用CountDown。

使用CountDownLatch

package com.example.demo.thread;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StopWatch;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 使用CountDown处理多个请求
 * @Author: dingguo
 * @Date: 2019/6/17 晚上8:23
 */
public class CountDownWorker {

    private static final Logger logger = LoggerFactory.getLogger(CountDownWorker.class);

    /**
     * 使用CountDownLatch批处理
     *
     * @param request
     * @return
     */
    private List<String> countDownDealTask(List<String> request) {
        StopWatch watch = new StopWatch();
        watch.start();

        //创建线程安全的返回结果
        List<String> result = Collections.synchronizedList(Lists.newArrayList());

        if (request != null) {
            //初始化计数器
            CountDownLatch countDownLatch = new CountDownLatch(request.size());
            //遍历处理
            request.forEach(s -> {
                try {
                    DealWorker worker = new DealWorker(countDownLatch, s, result);
                    //创建线程
                    new Thread(worker).start();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            //如果执行的过程中出现异常，则继续执行
            try {
                countDownLatch.await(1, TimeUnit.MINUTES);
            } catch (InterruptedException e) {
                logger.info("countDown deal task error[{}]", e.getMessage());
                e.printStackTrace();
            }
        }
        watch.stop();
        logger.info("花费时间:[{}]s", watch.getTotalTimeMillis());
        return result;
    }


    public static void main(String[] args) {
        CountDownWorker worker = new CountDownWorker();
        List<String> list = Lists.newArrayList("第一次请求", "第二次请求", "第三次请求");
        List<String> result = worker.countDownDealTask(list);
        //会使用批量获取的结果值
        logger.info("处理批量返回的数据：{}", result);
    }


    /**
     * 处理结果的线程
     */
    private class DealWorker implements Runnable {
        //计数器
        private CountDownLatch countDownLatch;
        //处理的请求
        private String detailRequest;
        //返回的结果集
        private List<String> result;

        public DealWorker(CountDownLatch countDownLatch, String detailRequest, List<String> result) {
            this.countDownLatch = countDownLatch;
            this.detailRequest = detailRequest;
            this.result = result;
        }

        @Override
        public void run() {
            try {
                //模拟耗时1s
                Thread.sleep(1000);
                result.add("处理线程:" + Thread.currentThread().getName() + "," + detailRequest);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //及时进行计数
                countDownLatch.countDown();
            }
        }
    }
}

执行结果：分别由三个线程执行，速度缩减为1s

使用了三个线程分别处理，效率变得更高。

注意事项

1. 各个线程的处理没有先后顺序，并发情况下无法保证顺序执行
2. 返回结果的列表是线程安全的
3. 各子线程的countDown计数确保要一定执行，否则会一直出现阻塞现象
4. 主线程的await要设置超时等待，出现异常及时抛出或者做相应的处理
5. 如果批量处理的请求过多，采用线程池，避免浪费线程资源。

参考文档

https://www.jianshu.com/p/9fc46f1f46b3

http://www.itzhai.com/the-introduction-and-use-of-a-countdownlatch.html