Jdk并发线程处理CountDownLatch、CyclicBa

作者: 水煮鱼又失败了 | 来源:发表于2020-05-20 08:21 被阅读0次

    目录

    [TOC]

    Java并发包提供了三个类,处理并发线程:CountdownLatch、Semaphore、CyclicBarrier。

    1. CountDownLatch

    1.1 描述

    • 中文含义"门栓",当指定数量的线程全部执行完毕后,主线程才继续向后执行

    • countDownLatch是一个一次性计数器,初始化需设置一个计数器的值。

    • 调用await方法可阻塞当前线程。

    • 调用一次countDown方法,计数器减1,当计数器减到0后,await方法阻塞的线程,继续向下执行。

    1.2 业务场景

    某个业务场景,程序需要分成很多线程一起执行,当全部子任务执行完成后(每个任务执行完,执行一次countDown,让计数器减1),再继续执行主程序后续的操作。

    如并行计算,处理的数据量很大,将任务拆分成多个子任务,等所有子任务都完成后,父任务再拿到所有自认为的运算结果进行汇总。

    1.3 代码

    public static void main(String[] args) throws InterruptedException {
        AtomicInteger result=new AtomicInteger(0);
        int count=5;
        CountDownLatch countDownLatch=new CountDownLatch(5);
        for(int i=0;i<count;i++){
            new Thread(() -> {
                try {
                    //业务逻辑...
                    result.addAndGet(10);
                    System.out.println(Thread.currentThread());
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    //保证计数器能顺利减1,否则父线程将一直阻塞
                    countDownLatch.countDown();
                }
            }).start();
        }
        //阻塞,等待计数器减为0
        countDownLatch.await();
        System.out.println("执行结果:"+result);
    }
    

    执行结果:

    Thread[Thread-0,5,main]
    Thread[Thread-1,5,main]
    Thread[Thread-2,5,main]
    Thread[Thread-3,5,main]
    Thread[Thread-4,5,main]
    执行结果:50
    

    2. CyclicBarrier

    2.1 说明

    • 中文含义"栅栏",多个线程互相等待,只要有一个线程没完成(没到屏障点),所有线程都等待
    • 调用await方法的线程告诉CyclicBarrier自己已经到达同步点,然后当前线程被阻塞。直到parties个参与线程调用了await方法,则全部阻塞的线程一起执行阻塞后面的代码。
    • 父线程不会阻塞。

    核心方法:

    • 等待:await()

    2.2 应用

    2.3 代码

    public class CyclicBarrierTest {
        /**
         * 执行数量5,配置执行后回调线程
         */
        public static CyclicBarrier cyclicBarrier=new CyclicBarrier(5,() -> {
            System.out.println("----------满足要求 call back thread----------");
        });
        
        public static void main(String[] args) throws Exception{
            for(int i=0;i<5;i++){
                new Thread(()->{
                    try {
                        System.out.println("进入线程:"+Thread.currentThread());
                        Thread.sleep(Long.valueOf(new Random().nextInt(3000)));
                        System.out.println("线程前置方法执行结束:"+Thread.currentThread());
                        //await会导致当前线程阻塞,当调用次数达到设置的值时,所有阻塞的线程将全部执行
                        cyclicBarrier.await();
                        System.out.println("满足条件,离开线程:"+Thread.currentThread());
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }).start();
            }
            //父线程并不会阻塞
            System.out.println("主线程结束");
        }
    }
    

    执行结果:

    主线程结束
    进入线程:Thread[Thread-3,5,main]
    进入线程:Thread[Thread-0,5,main]
    进入线程:Thread[Thread-2,5,main]
    进入线程:Thread[Thread-1,5,main]
    进入线程:Thread[Thread-4,5,main]
    线程前置方法执行结束:Thread[Thread-1,5,main]
    线程前置方法执行结束:Thread[Thread-4,5,main]
    线程前置方法执行结束:Thread[Thread-0,5,main]
    线程前置方法执行结束:Thread[Thread-3,5,main]
    线程前置方法执行结束:Thread[Thread-2,5,main]
    ----------满足要求 call back thread----------
    满足条件,离开线程:Thread[Thread-2,5,main]
    满足条件,离开线程:Thread[Thread-1,5,main]
    满足条件,离开线程:Thread[Thread-4,5,main]
    满足条件,离开线程:Thread[Thread-0,5,main]
    满足条件,离开线程:Thread[Thread-3,5,main]
    

    3. Semaphore

    3.1 说明

    • 中文含义”信号量“,保证同一批并发线程,最多同时执行的线程数,防止同时执行的线程过多

    • 限制最多可以有几个线程进来,超过最大线程数,新进来的线程可以选择等待,或者离开,如执行中的线程有结束的,则等待中的线程会自动加入到执行队列中,执行后续操作。

    • 可一次获取多个许可

    核心方法:

    • 获取权限:acquire()

    • 释放权限:release()

    3.2 应用

    特别适用于要求一个并发任务最多有几个线程执行的情况。

    3.3 代码

    3.1.1 获取一个许可
    public static void main(String[] args) {
        //最多允许2个线程执行
        Semaphore semaphore=new Semaphore(2);
        for(int i=0;i<3;i++){
            new Thread(()->{
                try {
                    //获取许可
                    semaphore.acquire();
                    //执行业务操作...
                    System.out.println(Thread.currentThread()+"----拿到许可"+LocalTime.now()); 
                    Thread.sleep(Long.valueOf(new Random().nextInt(3000)));
                    //释放许可
                    semaphore.release();
                    System.out.println(Thread.currentThread()+"----释放许可"+LocalTime.now()); 
                }catch (Exception e){
                    e.printStackTrace();
                }
            }).start();
        }
        System.out.println("执行结束");
    }
    

    执行结果:

    执行结束
    Thread[Thread-0,5,main]----拿到许可13:12:49.252
    Thread[Thread-1,5,main]----拿到许可13:12:49.252
    Thread[Thread-1,5,main]----释放许可13:12:49.906
    Thread[Thread-2,5,main]----拿到许可13:12:49.906
    Thread[Thread-2,5,main]----释放许可13:12:49.980
    Thread[Thread-0,5,main]----释放许可13:12:51.535
    
    3.1.2 获取多个许可
    public static void main(String[] args) {
        //最多允许4个线程执行
        Semaphore semaphore=new Semaphore(4);
        for(int i=0;i<3;i++){
            new Thread(()->{
                try {
                    //获取两个许可
                    semaphore.acquire(2);
                    //执行业务操作...
                    System.out.println(Thread.currentThread()+"----拿到许可"+LocalTime.now()); 
                    Thread.sleep(Long.valueOf(new Random().nextInt(3000)));
                    //释放两个许可
                    semaphore.release(2);
                    System.out.println(Thread.currentThread()+"----释放许可"+LocalTime.now()); 
                }catch (Exception e){
                    e.printStackTrace();
                }
            }).start();
        }
        System.out.println("执行结束");
    }
    

    执行结果:

    执行结束
    Thread[Thread-1,5,main]----拿到许可13:10:32.917
    Thread[Thread-0,5,main]----拿到许可13:10:32.917
    Thread[Thread-1,5,main]----释放许可13:10:33.208
    Thread[Thread-2,5,main]----拿到许可13:10:33.208
    Thread[Thread-2,5,main]----释放许可13:10:35.031
    Thread[Thread-0,5,main]----释放许可13:10:35.491
    
    3.1.3 尝试获取许可
    public static void main(String[] args) {
        //最多允许2个线程执行
        Semaphore semaphore=new Semaphore(2);
        for(int i=0;i<5;i++){
            new Thread(()->{
                try {
                    //获取许可
                    if(semaphore.tryAcquire()){
                        //执行业务操作...
                        System.out.println(Thread.currentThread()+"----拿到许可"+LocalTime.now()); 
                        Thread.sleep(100);
                        //释放许可
                        semaphore.release();
                        System.out.println(Thread.currentThread()+"----释放许可"+LocalTime.now()); 
                    }else{
                        System.out.println(Thread.currentThread()+"----"+"未获取许可,放弃"+LocalTime.now());
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }).start();
        }
        System.out.println("执行结束");
    }
    

    执行结果:

    执行结束
    Thread[Thread-4,5,main]----未获取许可,放弃13:22:01.626
    Thread[Thread-2,5,main]----未获取许可,放弃13:22:01.626
    Thread[Thread-1,5,main]----拿到许可13:22:01.626
    Thread[Thread-3,5,main]----未获取许可,放弃13:22:01.626
    Thread[Thread-0,5,main]----拿到许可13:22:01.626
    Thread[Thread-0,5,main]----释放许可13:22:01.727
    Thread[Thread-1,5,main]----释放许可13:22:01.727
    
    3.1.4 指定时间内尝试获取许可
    public static void main(String[] args) {
        //最多允许2个线程执行
        Semaphore semaphore=new Semaphore(2);
        for(int i=0;i<5;i++){
            new Thread(()->{
                try {
                    //获取许可
                    if(semaphore.tryAcquire(1, TimeUnit.SECONDS)){
                        //执行业务操作...
                        System.out.println(Thread.currentThread()+"----"+LocalTime.now()+"----拿到许可"); 
                        Thread.sleep(Long.valueOf(new Random().nextInt(3000)));
                        //释放许可
                        semaphore.release();
                        System.out.println(Thread.currentThread()+"----"+LocalTime.now()+"----释放许可"); 
                    }else{
                        System.out.println(Thread.currentThread()+"----"+LocalTime.now()+"----1秒内未获取许可,放弃");
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }).start();
        }
        System.out.println("执行结束");
    }
    

    执行结果:

    执行结束
    Thread[Thread-3,5,main]----13:24:29.752----拿到许可
    Thread[Thread-2,5,main]----13:24:29.752----拿到许可
    Thread[Thread-2,5,main]----13:24:30.088----释放许可
    Thread[Thread-0,5,main]----13:24:30.088----拿到许可
    Thread[Thread-4,5,main]----13:24:30.732----1秒内未获取许可,放弃
    Thread[Thread-1,5,main]----13:24:30.733----1秒内未获取许可,放弃
    Thread[Thread-3,5,main]----13:24:31.929----释放许可
    Thread[Thread-0,5,main]----13:24:32.340----释放许可
    

    4. 补充(引用其他网友)

    • CountDownLatch和CyclicBarrier区别:

      CountDownLatch和CyclicBarrier都有让多个线程等待同步然后再开始下一步动作的意思,但是CountDownLatch的下一步的动作实施者是主线程,具有不可重复性;
      而CyclicBarrier的下一步动作实施者还是“其他线程”本身

    相关文章

      网友评论

        本文标题:Jdk并发线程处理CountDownLatch、CyclicBa

        本文链接:https://www.haomeiwen.com/subject/uixbohtx.html