Java并发工具类

迈不过友情╰ 2022-05-08 13:24 286阅读 0赞

在Java1.5中，提供了一些很有用的辅助类来帮助我们进行并发编程，比如CountDownLatch\[计数器\],CyclicBarrier\[循环屏障\]，Semaphore，下面对三个类进行讲解

一、CountDownLatch用法：属于java.util.concurrent包

CountDownLatch是一个同步工具类，用来协调多个线程之间的同步，或者说起到线程之间的通信（而不是用作互斥的作用）

原理：CountDownLatch类实际上是使用计数器的方式去控制的，不难想象当我们初始化CountDownLatch的时候传入了一个int变量这个时候在类的内部初始化一个int的变量，每当我们调用countDownt()方法的时候就使得这个变量的值减1，而对于await()方法则去判断这个int的变量的值是否为0，是则表示所有的操作都已经完成，否则继续等待

使用场景：

1）某一个线程在运行前等待n个线程执行完毕，将CountDownLatch的计数器初始化为n，即new CountDownLatch(n),每当一个线程完成工作，计数器减一，即调用countDown（）方法，当计数器数值变为0时，CountDownLatch上的await（）的线程就会被唤醒。典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行

2）实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计数器初始化为1，多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为0，多个线程同时被唤醒

场景一演示：

package com.multithread.object;
    
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * CountDownLatchDemo的使用
     */
    public class CountDownLatchDemo {
    
        public static void main(String[] args) {
            final CountDownLatch latch = new CountDownLatch(3);
            for (int i = 0; i < 3; i++) {
                new Thread() {
                    @Override
                    public void run() {
                        try {
                            System.out.println("子线程：" + Thread.currentThread().getName() + "正在执行");
                            Thread.sleep(3000);
                            System.out.println("子线程:" + Thread.currentThread().getName() + "执行完毕");
                            //执行完成，减一
                            latch.countDown();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }.start();
            }
    
            try{
                System.out.println("等待3个子线程执行完毕...");
                //等待
                latch.await();
                System.out.println("3个子线程执行完毕");
                System.out.println("继续执行主线程");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
        }
    
    }

执行结果如下：

等待3个子线程执行完毕...  
子线程：Thread-0正在执行  
子线程：Thread-1正在执行  
子线程：Thread-2正在执行  
子线程:Thread-0执行完毕  
子线程:Thread-1执行完毕  
子线程:Thread-2执行完毕  
3个子线程执行完毕  
继续执行主线程

场景二演示：

package com.multithread.object;
    
    import java.util.concurrent.CountDownLatch;
    
    public class MyCountDownLatchDemo implements Runnable {
    
        private final CountDownLatch countDownLatch;
        private final CountDownLatch latch;
    
        public MyCountDownLatchDemo(CountDownLatch countDownLatch, CountDownLatch latch) {
            this.countDownLatch = countDownLatch;
            this.latch = latch;
        }
    
        public static void main(String[] args) throws InterruptedException {
            CountDownLatch countDownLatch = new CountDownLatch(1);
            CountDownLatch latch = new CountDownLatch(6);
    
            //创建并启动6个处于等待状态的MyCountDownLatchDemo线程
            for (int i = 0; i < 6; i++) {
                new Thread(new MyCountDownLatchDemo(countDownLatch,latch)).start();
            }
    
            System.out.println("用于触发处于等待状态的线程开始工作......");
            System.out.println("用于触发处于等待状态的线程工作完成，等待状态线程开始工作......");
            countDownLatch.countDown();//主线程执行了这边一步之后，countDownLatch计数器为0，子线程的run方法的业务逻辑才开始执行，所有的子线程同时执行，统一起跑线
            latch.await();
            System.out.println("Bingo!");
        }
    
        @Override
        public void run() {
            try{
                //countDownLatch计数器为1，等待主线程执行完毕，子线程再执行
                countDownLatch.await();
                //-----
                /*业务逻辑*/
                //-----
                System.out.println("主线程执行完毕，子线程"+Thread.currentThread().getName()+"开始执行");
                latch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

结果:  
用于触发处于等待状态的线程开始工作......  
用于触发处于等待状态的线程工作完成，等待状态线程开始工作......  
主线程执行完毕，子线程Thread-0开始执行  
主线程执行完毕，子线程Thread-1开始执行  
主线程执行完毕，子线程Thread-2开始执行  
主线程执行完毕，子线程Thread-3开始执行  
主线程执行完毕，子线程Thread-4开始执行  
主线程执行完毕，子线程Thread-5开始执行  
Bingo!

二、CyclicBarrier ：位于java.util.concurrent包下

用法：可以实现让一组线程等待至某个状态之后再同步执行

简介：CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞

场景一：若干线程都要进行一个相同的操作，只有全部线程执行完毕之后，才能进行后续的事情，就可以利用CyclicBarrier

import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class MyCyclicBarrier {
        public static void main(String[] args) {
            int N = 4;
            CyclicBarrier barrier = new CyclicBarrier(N);
            for (int i = 0; i < N; i++){
                new DoSomething(barrier).start();
            }
        }
    
        static class DoSomething extends Thread {
            private CyclicBarrier cyclicBarrier;
    
            public DoSomething(CyclicBarrier cyclicBarrier) {
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() {
                System.out.println("线程" + Thread.currentThread().getName() + "正在做.....");
                try {
                    Thread.sleep(5000);      //以睡眠来模拟业务逻辑操作
                    System.out.println("线程" + Thread.currentThread().getName() + "doSomething完毕，等待其他线程完成");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
                System.out.println("所有线程dosomething完毕，继续处理其他任务...");
            }
        }
    }

执行结果：

线程Thread-0正在做.....  
线程Thread-1正在做.....  
线程Thread-2正在做.....  
线程Thread-3正在做.....  
线程Thread-0doSomething完毕，等待其他线程完成  
线程Thread-1doSomething完毕，等待其他线程完成  
线程Thread-2doSomething完毕，等待其他线程完成  
线程Thread-3doSomething完毕，等待其他线程完成  
所有线程dosomething完毕，继续处理其他任务...  
所有线程dosomething完毕，继续处理其他任务...  
所有线程dosomething完毕，继续处理其他任务...  
所有线程dosomething完毕，继续处理其他任务...

进阶：如果说想在所有线程写入操作完之后，进行额外的其他操作可以为CyclicBarrier提供Runnable参数

package com.multithread.object;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class MyCyclicBarrierDemo2 {
        public static void main(String[] args) {
            int N = 4;
            CyclicBarrier barrier  = new CyclicBarrier(N,new Runnable() {
                @Override
                public void run() {
                    System.out.println("当前线程"+Thread.currentThread().getName());
                }
            });
    
            for(int i=0;i<N;i++)
                new DoSomething(barrier).start();
        }
        static class DoSomething extends Thread{
            private CyclicBarrier cyclicBarrier;
            public DoSomething(CyclicBarrier cyclicBarrier) {
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() {
                System.out.println("线程"+Thread.currentThread().getName()+"正在dosomething...");
                try {
                    Thread.sleep(5000);      //以睡眠来模拟业务逻辑操作
                    System.out.println("线程"+Thread.currentThread().getName()+"doSomething完毕，等待其他线程写入完毕");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }catch(BrokenBarrierException e){
                    e.printStackTrace();
                }
                System.out.println("所有线程dosomething完毕，继续处理其他任务...");
            }
        }
    }

结果：

线程Thread-1正在dosomething...  
线程Thread-0正在dosomething...  
线程Thread-3正在dosomething...  
线程Thread-2正在dosomething...  
线程Thread-3doSomething完毕，等待其他线程写入完毕  
线程Thread-0doSomething完毕，等待其他线程写入完毕  
线程Thread-2doSomething完毕，等待其他线程写入完毕  
线程Thread-1doSomething完毕，等待其他线程写入完毕  
当前线程Thread-2  
所有线程dosomething完毕，继续处理其他任务...  
所有线程dosomething完毕，继续处理其他任务...  
所有线程dosomething完毕，继续处理其他任务...  
所有线程dosomething完毕，继续处理其他任务...  
可以看到：当4个子线程都完成任务之后，在子线程中随机选择一个线程执行runnable里面的方法

### CyclicBarrier和CountDownLatch的区别： ###

CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置

CyclicBarrier能处理更为复杂的业务场景，比如如果计算发生错误，可以重置计数器，并让线程们重新执行一次

CyclicBarrier还提供其他有用的方法，比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。比如以下代码执行完之后会返回true

三、Semaphore 字面意思信号量，可以控制同时访问线程的个数通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可，位于java.util.concurrent包下

1.Semaphore的构造器

/**
         * Creates a {@code Semaphore} with the given number of
         * permits and nonfair fairness setting.
         *
         * @param permits the initial number of permits available.
         *        This value may be negative, in which case releases
         *        must occur before any acquires will be granted.
         */
    //参数permits表示许可数目，即同时可以允许多少线程进行访问
        public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
    
        /**
         * Creates a {@code Semaphore} with the given number of
         * permits and the given fairness setting.
         *
         * @param permits the initial number of permits available.
         *        This value may be negative, in which case releases
         *        must occur before any acquires will be granted.
         * @param fair {@code true} if this semaphore will guarantee
         *        first-in first-out granting of permits under contention,
         *        else {@code false}
         */
    /这个多了一个参数fair表示是否是公平的，即等待时间越久的越先获取许可
        public Semaphore(int permits, boolean fair) {
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

2.重要的方法

`public` `void` `acquire()``throws` `InterruptedException {  }    ``//获取一个许可`

`public` `void` `acquire(``int` `permits)``throws` `InterruptedException { }   ``//获取permits个许可`

`public` `void` `release() { }         ``//释放一个许可`

`public` `void` `release(``int` `permits) { }   ``//释放permits个许可`

acquire()用来获取一个许可，若无许可能够获得，则会一直等待，直到获得许可。

release()用来释放许可。注意，在释放许可之前，必须先获获得许可

上述4个方法都会被阻塞，如果想立即得到执行结果，可以使用下面几个方法：

`public` `boolean` `tryAcquire() { };   ``//尝试获取一个许可，若获取成功，则立即返回true，若获取失败，则立即返回false`

`public` `boolean` `tryAcquire(``long` `timeout, TimeUnit unit)``throws` `InterruptedException { }; ``//尝试获取一个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false`

`public` `boolean` `tryAcquire(``int` `permits) { };``//尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false`

`public` `boolean` `tryAcquire(``int` `permits,``long` `timeout, TimeUnit unit)``throws` `InterruptedException { };``//尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false`

另外还可以通过availablePermits()方法得到可用的许可数目

package com.multithread.object;
    
    import java.util.concurrent.Semaphore;
    
    public class SemaphereDemo {
        public static void main(String[] args) {
            int N = 8;            //玩家数
            Semaphore semaphore = new Semaphore(5); //电脑数目
            for(int i=0;i<N;i++)
                new Computer(i,semaphore).start();
        }
    
        static class Computer extends Thread{
            private int num;
            private Semaphore semaphore;
            public Computer(int num,Semaphore semaphore){
                this.num = num;
                this.semaphore = semaphore;
            }
    
            @Override
            public void run() {
                try {
                    semaphore.acquire();
                    System.out.println("玩家"+this.num+"占用一个电脑在玩游戏...");
                    Thread.sleep(2000);
                    System.out.println("玩家"+this.num+"释放出电脑");
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

结果：

玩家0占用一个电脑在玩游戏...  
玩家1占用一个电脑在玩游戏...  
玩家2占用一个电脑在玩游戏...  
玩家3占用一个电脑在玩游戏...  
玩家4占用一个电脑在玩游戏...  
玩家0释放出电脑  
玩家5占用一个电脑在玩游戏...  
玩家1释放出电脑  
玩家6占用一个电脑在玩游戏...  
玩家2释放出电脑  
玩家7占用一个电脑在玩游戏...  
玩家3释放出电脑  
玩家4释放出电脑  
玩家5释放出电脑  
玩家6释放出电脑  
玩家7释放出电脑