Jdk并发线程处理CountDownLatch、CyclicBa

2020-05-20 本文已影响0人水煮鱼又失败了

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

目录

1. CountDownLatch

1.1 描述

1.2 业务场景

1.3 代码

2. CyclicBarrier

2.1 说明

2.2 应用

2.3 代码

3. Semaphore

3.1 说明

3.2 应用

3.3 代码

3.1.1 获取一个许可

3.1.2 获取多个许可

3.1.3 尝试获取许可

3.1.4 指定时间内尝试获取许可

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

猜你喜欢

热点阅读