并发编程之CountDownLatch&Semaphore&CyclicBarrier&Executors

2022-06-16 12:18:52

一、Semaphore【sei mo fo】

Semaphore 定义

Semaphore 字面意思是信号量的意思，它的作用是控制访问特定资源的线程数目，底层依赖AQS的状态State，是在生产当中比较常用的一个工具类。

Semaphore常用方法

构造方法

// 设置reentrantLock 的state个数为 permists个，意味着最多有permists个线程同时竞争锁资源，其他的线程被放入到CLH队列中publicSemaphore(intpermits)// permits代表初始state的值，代表了系统中所允许的最大并行度publicSemaphore(intpermits,boolean fair)
  • permits 表示许可线程的数量
  • fair 表示公平性，如果这个设为 true 的话，下次执行的线程会是等待最久的线程

常用方法

// 支持中断，线程被中断会抛出异常publicvoidacquire()throwsInterruptedException// 从总的池子里拿出一个凭证（state = state -1）acquire(intpermits)// state = state - permitstryAcquire(long timeout,TimeUnit unit)// 设置获取时的超时等待时间// 以上的方法都支持中断，但是下面的方法不支持中断acquireUninterruptibly()// 非中断的方式竞争锁，不支持中断// 我们自己使用的时候最好使用可以中断的，一些线程如果不想让其继续竞争，可以中断==================================================================publicvoidrelease()// 将获取的的凭证放回到池子里， state  = state + 1release(intpermits)// state  = state + permits

tryAcquire（long timeout,TimeUnit unit）
  • acquire() 表示阻塞并获取许可
  • release() 表示释放许可

基本使用

使用场景：
资源访问，服务限流(Hystrix里限流就有基于信号量方式)。

代码实现：

packagecom.jihu.test.semaphore;importjava.util.concurrent.Semaphore;publicclassSemaphoreRunner{publicstaticvoidmain(String[] args){Semaphore semaphore=newSemaphore(2);for(int i=0; i<5; i++){newThread(newTask(semaphore,"xiao yan+"+ i)).start();}}staticclassTaskextendsThread{Semaphore semaphore;publicTask(Semaphore semaphore,String tname){this.semaphore= semaphore;this.setName(tname);}publicvoidrun(){try{
                semaphore.acquire();System.out.println(Thread.currentThread().getName()+":aquire() at time:"+System.currentTimeMillis());Thread.sleep(1000);
                semaphore.release();System.out.println(Thread.currentThread().getName()+":aquire() at time:"+System.currentTimeMillis());}catch(InterruptedException e){
                e.printStackTrace();}}}}

结果：

Thread-1:aquire() at time:1622590044923Thread-3:aquire() at time:1622590044924Thread-3:aquire() at time:1622590045931Thread-5:aquire() at time:1622590045931Thread-7:aquire() at time:1622590045932Thread-1:aquire() at time:1622590045932Thread-9:aquire() at time:1622590046931Thread-5:aquire() at time:1622590046931Thread-7:aquire() at time:1622590046932Thread-9:aquire() at time:1622590047932Process finishedwithexit code0

从打印结果可以看出，一次只有两个线程执行 acquire()，只有线程进行 release() 方法后才会有别的线程执行 acquire()。

我们再来看一个类似限流降级的例子：

packagecom.jihu.test.semaphore;importjava.util.concurrent.Semaphore;importjava.util.concurrent.TimeUnit;publicclassSemaphoreTest1{publicstaticvoidmain(String[] args){Semaphore semaphore=newSemaphore(2);for(int i=0; i<10; i++){newThread(newTask(semaphore,"xiaoyan")).start();}}staticclassTaskextendsThread{Semaphore semaphore;publicTask(Semaphore semaphore,String tname){super(tname);this.semaphore= semaphore;}@Overridepublicvoidrun(){try{// 如果只有两个线程在特定的时间获取到锁，否则服务就会被降级if(semaphore.tryAcquire(500,TimeUnit.MILLISECONDS)){System.out.println(Thread.currentThread().getName()+":acquire() at time "+System.currentTimeMillis());Thread.sleep(5000);
                    semaphore.release();}else{fallback();}}catch(InterruptedException e){
                e.printStackTrace();}}privatevoidfallback(){System.out.println("服务降级！");}}}

结果：

Thread-0:acquire() at time1622601697525Thread-1:acquire() at time1622601697527
服务降级！
服务降级！
服务降级！
服务降级！
服务降级！
服务降级！
服务降级！
服务降级！Process finishedwithexit code0

AQS的共享模式

state 是一个带volatile前缀的int值,是一个类似计数器的东西。在不同的同步组件中有不同的含义。以ReentrantLock为例,state可以用来表示该锁被线程重入的次数。当state为0表示该锁不被任何线程持有;当state为1表示线程恰好持有该锁1次(未重入);当state大于1则表示锁被线程重入state次。因为这是一个会被并发访问的量,为了防止出现可见性问题要用volatile进行修饰。

独占模式：之前我们看的reentrantLock是AQS的独占模式，即state = 0的时候，线程可以竞争锁资源；state > 0的时候，线程需要入队阻塞；

共享模式：当state = n( n> 0)的时候，可以有n个线程来竞争锁资源。

二、CountDownLatch

定义

CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。等待线程池中所有线程都执行完成后在结束。

使用场景：
Zookeeper分布式锁,Jmeter模拟高并发等。

工作原理

CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

常用API

CountDownLatch.countDown()// 等待计数器中线程数减1CountDownLatch.await();// 等待计数器中所有线程都执行完成==============  源码==============publicvoidcountDown(){
        sync.releaseShared(1);}

应用场景举例

比如陪媳妇去看病。

医院里边排队的人很多，如果一个人的话，要先看大夫，看完大夫再去排队交钱取药。现在我们是双核，可以同时做这两个事（多线程）。假设看大夫花3秒钟，排队交费取药花5秒钟。我们同时搞的话，5秒钟我们就能完成，然后一起回家（回到主线程）。

代码如下：

/**
 * 看大夫任务
 * */publicclassSeeDoctorTaskimplementsRunnable{privateCountDownLatch countDownLatch;publicSeeDoctorTask(CountDownLatch countDownLatch){this.countDownLatch= countDownLatch;}publicvoidrun(){try{System.out.println("开始看医生");Thread.sleep(3000);System.out.println("看医生结束，准备离开病房");}catch(InterruptedException e){
            e.printStackTrace();}finally{if(countDownLatch!=null)
                countDownLatch.countDown();}}}
/**
 * 排队的任务
 */publicclassQueueTaskimplementsRunnable{privateCountDownLatch countDownLatch;publicQueueTask(CountDownLatch countDownLatch){this.countDownLatch= countDownLatch;}publicvoidrun(){try{System.out.println("开始在医院药房排队买药....");Thread.sleep(5000);System.out.println("排队成功，可以开始缴费买药");}catch(InterruptedException e){
            e.printStackTrace();}finally{if(countDownLatch!=null)
                countDownLatch.countDown();}}}
/**
 * 配媳妇去看病，轮到媳妇看大夫时
 * 我就开始去排队准备交钱了。
 */publicclassCountDownLaunchRunner{publicstaticvoidmain(String[] args)throwsInterruptedException{long now=System.currentTimeMillis();CountDownLatch countDownLatch=newCountDownLatch(2);newThread(newSeeDoctorTask(countDownLatch)).start();newThread(newQueueTask(countDownLatch)).start();//等待线程池中的2个任务执行完毕，否则一直等待
        countDownLatch.await();long time=System.currentTimeMillis()- now;System.out.println("over，回家 cost:"+ time);}}

三、CyclicBarrier

栅栏屏障，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

CyclicBarrier默认的构造方法是CyclicBarrier（int parties），其参数表示屏障拦截的线程数量，每个线程调用await方法告CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。

常见方法

cyclicBarrier.await();

应用场景

可以用于多线程计算数据，最后合并计算结果的场景。例如，用一个Excel保存了用户所有银行流水，每个Sheet保存一个账户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完之后，得到每个sheet的日均银行流水，最后，再用barrierAction用这些线程的计算结果，计算出整个Excel的日均银行流水。示例代码：

publicclassCyclicBarrierRunnerimplementsRunnable{privateCyclicBarrier cyclicBarrier;privateint index;publicCyclicBarrierRunner(CyclicBarrier cyclicBarrier,int index){this.cyclicBarrier= cyclicBarrier;this.index= index;}publicvoidrun(){try{System.out.println("index: "+ index);
            index--;
            cyclicBarrier.await();}catch(Exception e){
            e.printStackTrace();}}publicstaticvoidmain(String[] args)throwsException{CyclicBarrier cyclicBarrier=newCyclicBarrier(11,newRunnable(){publicvoidrun(){System.out.println("所有特工到达屏障，准备开始执行秘密任务");}});for(int i=0; i<10; i++){newThread(newCyclicBarrierRunner(cyclicBarrier, i)).start();}
        cyclicBarrier.await();System.out.println("全部到达屏障....");//  多次使用for(int i=0; i<10; i++){newThread(newCyclicBarrierRunner(cyclicBarrier, i)).start();}
        cyclicBarrier.await();System.out.println("全部到达屏障....");}}

结果：

index:0
index:1
index:2
index:3
index:4
index:5
index:6
index:7
index:8
index:9
所有特工到达屏障，准备开始执行秘密任务
全部到达屏障....Process finishedwithexit code0

CyclicBarrier可以执行都次，但是countdownlatch只能一次。

四、Executors

主要用来创建线程池，代理了线程池的创建，使得你的创建入口参数变得简单。

和线程池相关的可以看这篇文章：Java并发ThreadPoolExecutor原理源码详解 - https://blog.csdn.net/qq_43631716/article/details/114788744

重要方法

  • newCachedThreadPool：创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
  • newFixedThreadPool：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
  • newScheduledThreadPool ：创建一个定长线程池， 支持定时及周期性任务执行。
  • newSingleThreadExecutor创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

缺点（阿里不推荐使用）：

Executors创建出来的线程池使用的全都是无界队列，而使用无界队列会带来很多弊端，最重要的就是，它可以无限保存任务，因此很有可能造成OOM异常。同时在某些类型的线程池里面，使用无界队列还会导致maxinumPoolSize、keepAliveTime、handler等参数失效.

CachedThreadPool: 这种最大会创建Integer.MAX_VALUE个线程，会消耗大量的CPU资源。

五、Exchanger

当一个线程运行到exchange()方法时会阻塞，另一个线程运行到exchange()时，二者交换数据，然后执行后面的程序。

应用场景极少，大家了解即可。

packagecom.jihu.test.exchanger;importjava.util.concurrent.Exchanger;publicclassExchangerTest{publicstaticvoidmain(String[] args){finalExchanger<Integer> exchanger=newExchanger<Integer>();for(int i=0; i<10; i++){finalInteger num= i;newThread(){publicvoidrun(){System.out.println("我是线程：Thread_"+this.getName()+"我的数据是："+ num);try{Integer exchangeNum= exchanger.exchange(num);Thread.sleep(1000);System.out.println("我是线程：Thread_"+this.getName()+"我原先的数据为："+ num+" , 交换后的数据为："+ exchangeNum);}catch(InterruptedException e){
                        e.printStackTrace();}}}.start();}}}
  • 作者：Jihu Tuo
  • 原文链接：https://blog.csdn.net/qq_43631716/article/details/117458767
    更新时间：2022-06-16 12:18:52