CountDownLatch

表示计数器，可以给CountDownLatch设置一个数字

一个线程调用CountDownLatch的await()方法，将会阻塞

其他线程可以调用CountDownLatch的countDown()方法来对CountDownLatch中的数字减一， 当数字被减成0后，所有await的线程都将被唤醒。

对应的底层原理就是，调用await()方法的线程会利用AQS排队，一旦数字被减为0，则会将AQS中排队的线程依次唤醒。

Semaphore

Semaphore就是信号量,Semaphore可以阻塞线程并且可以控制同时访问线程的个数,通过acquire()获取一个许可,如果没有获取到就继续等待,通过release()释放一个许可。Semaphore和锁有点类似,都可以控制对某个资源的访问权限。

使用场景

CountDownLatch可以实现类似计数器的功能。比如有一个任务A,它要等到其它3任务完成才能执行,此时就可以用CountDownLatch来实现。

Semaphore可以用来做流量分流，特别是对公共资源有限的场景，比如数据库连接。
假设有这个的需求，读取几万个文件的数据到数据库中，由于文件读取是IO密集型任务，可以启动几十个线程并发读取，但是数据库连接数只有10个，这时就必须控制最多只有10个线程能够拿到数据库连接进行操作。这个时候，就可以使用Semaphore做流量控制。

案例1：多线程分段处理List集合

实际应用中，分批后线程数量过大，会导致线程阻塞，线程切换上下文，效率不高，本次是根据数据量动态设置线程数，同时控制最大并发数量（业务中有IO操作，避免过大并发导致堵塞），实现效率提高

思路：

         1根据数据量动态设置线程数
         2.使用Semaphore 控制允许并发访问线程的个数
         3.CountDownLatch计数器闭锁

如何控制某个方法允许并发访问线程的个数？
    Semaphore类有两个重要方法
    1、semaphore.acquire();
        请求一个信号量，这时候信号量个数-1，当减少到0的时候，下一次acquire不会再执行，只有当执行一个release()的时候，信号量不为0的时候才可以继续执行acquire
    2、semaphore.release();
        释放一个信号量，这时候信号量个数+1，

也就是说在线程里执行某个方法的时候，在方法里用该类对象进行控制，就能保证所有的线程中最多只有指定信号量个数个该方法在执行。

举例：我开启了100个线程，执行一个()方法，但是我只想要所有线程中，最多有五个线程在执行该方法，其他的线程就必须排队等待。
则可以使用Semaphore对象进行控制，该对象new初始化的时候有个int参数，即指定最多信号量个数。
 

  /**
     * <p>
     * 实例3：多线程分段处理List集合
     * </P>
     */
    @Test
    public void multiThreadedListHandler3() {

        // 开始时间
        long start = System.currentTimeMillis();
        List<String> list = new ArrayList<>(10000);
        for (long i = 1; i <= 20000000; i++) {
            list.add(i + "");
        }
        int listSize = list.size();
        //执行数量
        int limitNum = 1000;
        //线程数(也就是执行次数)
        int threadNum = listSize % limitNum == 0 ? listSize / limitNum : listSize / limitNum + 1;
        int pcount = Runtime.getRuntime().availableProcessors();
        //最大线程数控制
        int maxthreadNum = 5;

        ExecutorService executor = new ThreadPoolExecutor(5, maxthreadNum, 1, TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(1000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        //最大并发线程数控制
        final Semaphore semaphore = new Semaphore(maxthreadNum);
        List handleList = null;
        for (int i = 0; i < threadNum; i++) {
            if ((i + 1) == threadNum) {
                int startIndex = i * limitNum;
                int endIndex = list.size();
                handleList = list.subList(startIndex, endIndex);
            } else {
                int startIndex = i * limitNum;
                int endIndex = (i + 1) * limitNum;
                handleList = list.subList(startIndex, endIndex);
            }
            SyncTask task = new SyncTask(handleList, countDownLatch, semaphore);
            executor.execute(task);
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
            System.out.println("线程任务执行结束");
            System.err.println("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
        }
    }





@Slf4j
static class SyncTask implements Runnable {
    private List<String> list;
    private CountDownLatch countDownLatch;
    private Semaphore semaphore;

    public SyncTask(List<String> list, CountDownLatch countDownLatch, Semaphore semaphore) {
        this.list = list;
        this.countDownLatch = countDownLatch;
        this.semaphore = semaphore;
    }

    @Override
    public void run() {
        if (!CollectionUtils.isEmpty(list)) {
            try {
                semaphore.acquire();
                list.stream().forEach(t -> {
                    //业务处理
                });
                System.out.println(Thread.currentThread().getName() + "线程：" + list);
                //  log.debug(String.format("%s", Thread.currentThread().getName() + "线程：" + list));
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
        }
        //线程任务完成
        countDownLatch.countDown();
    }
}

案例2：

将集合按指定数量分组，list中的元素被平均分配到n个集合中(平均分配给坐席)

/**
     * 将集合按指定数量分组，list中的元素被平均分配到n个集合中(平均分配给坐席)
     *
     * @param list             数据集合
     * @param currentSeatsList 客服名单集合
     * @return 分组结果
     */
    @Test
    public void collectionElementsAreGroupedByAverage() {

        // 开始时间
        long start = System.currentTimeMillis();
        //数据
        List<String> list = new ArrayList<>(10);
        for (long i = 1; i <= 123; i++) {
            list.add(i + "");
        }
        //客服名单
        List<String> currentSeatsList = new ArrayList<>(10);
        for (long i = 1; i <= 12; i++) {
            currentSeatsList.add("customer" + i);
        }

        int listSize = list.size();
        //根据坐席名单计算分批数量,分批数量=listSize/坐席数量
        int limitNum = listSize % currentSeatsList.size() == 0 ? listSize / currentSeatsList.size() : listSize / currentSeatsList.size() + 1;
        //执行次数
        int batchNum = listSize % limitNum == 0 ? listSize / limitNum : listSize / limitNum + 1;
        //cpu
        int pcount = Runtime.getRuntime().availableProcessors();
        //最大线程数控制
        int maxthreadNum = 5;

        ExecutorService executor = new ThreadPoolExecutor(pcount, maxthreadNum, 1, TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(batchNum), Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

        //CountDownLatch计数器闭锁
        CountDownLatch countDownLatch = new CountDownLatch(batchNum);
        //控制最大并发线程数量
        final Semaphore semaphore = new Semaphore(maxthreadNum);

        List handleList = null;
        for (int i = 0; i < batchNum; i++) {
            if ((i + 1) == batchNum) {
                int startIndex = i * limitNum;
                int endIndex = list.size();
                handleList = list.subList(startIndex, endIndex);
            } else {
                int startIndex = i * limitNum;
                int endIndex = (i + 1) * limitNum;
                handleList = list.subList(startIndex, endIndex);
            }
            Task1 task = new Task1(handleList, countDownLatch, semaphore, currentSeatsList.get(i));
            executor.execute(task);
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
            System.out.println("线程任务执行结束");
            System.err.println("执行任务消耗了 ：" + (System.currentTimeMillis() - start) + "毫秒");
        }
    }



@Slf4j
class Task1 implements Runnable {
    private List<String> list;
    private CountDownLatch countDownLatch;
    private Semaphore semaphore;
    private String currentSeats;


    public Task1(List<String> list, CountDownLatch countDownLatch, Semaphore semaphore, String currentSeats) {
        this.list = list;
        this.countDownLatch = countDownLatch;
        this.semaphore = semaphore;
        this.currentSeats = currentSeats;
    }

    @Override
    public void run() {
        if (!CollectionUtils.isEmpty(list)) {
            try {
                semaphore.acquire();
                list.stream().forEach(t -> {
                    //业务处理
                });
                String strs = String.format("客服:%s  线程:%s", currentSeats, Thread.currentThread().getName() + list);
                System.out.println(strs);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
        }
        //线程任务完成
        countDownLatch.countDown();
    }

}