Java并发编程:JUC辅助工具类

2022-09-30 11:05:26

1. CountDownLatch:等待多线程完成

  • CountDownLatch 允许一个或多个线程等待其他线程完成操作。CountDownLatch 可以实现的功能和join()方法类似。
  • join() 方法用于让当前执行线程等待 join 线程执行结束。其实现原理是不停的检查 join 线程是否存活,如果 join 线程存活则让当前线程永远等待(wait(0);),直到 join 线程终止后,当前线程的this.notifyAll();方法才会被调用,调用 notifyAll() 方法是在 JVM 里实现的。
  • CountDownLatch 主要有两个方法:
    • 当一个或多个线程调用await()方法时,这些线程会被阻塞。
    • 其他线程调用countDown()方法会将计数器减1(调用countDown方法的线程不会阻塞)。当计数器的值变为0时,因await()方法被阻塞的线程会被唤醒,继续执行。
    • 如果有某个线程处理速度非常慢,为了避免一直等待,可以使用另一个带指定时间的await方法await(long timeout, TimeUnit unit),这个方法等待特定时间后,就不会阻塞当前线程了。
package com.java.juc1;import java.util.concurrent.CountDownLatch;/**
 * @author rrqstart
 * 案例:教室里一共有6个普通同学和1个班长,要求6个普通同学离开教室后班长才能锁门走人。
 */publicclassCountDownLatchTest{publicstaticvoidmain(String[] args){//CountDownLatch的构造器接收一个int类型的参数count作为计数器。//该计数器可以用于计数count个线程,也可以计数1个线程里的count个步骤。//计数器必须大于等于0,计数器等于0时,调用await()方法时不会阻塞当前线程。//CountDownLatch不能重新初始化,也不能修改CountDownLatch对象的内部计数器的值。
        CountDownLatch countDownLatch=newCountDownLatch(6);for(int i=0; i<6; i++){newThread(()->{
                System.out.println(Thread.currentThread().getName()+"号同学离开教室!");
                countDownLatch.countDown();}, String.valueOf(i+1)).start();}try{
            countDownLatch.await();}catch(InterruptedException e){
            e.printStackTrace();}

        System.out.println(Thread.currentThread().getName()+"班长最后锁门走人!");}}
//执行结果:1号同学离开教室!5号同学离开教室!4号同学离开教室!3号同学离开教室!2号同学离开教室!6号同学离开教室!
main班长最后锁门走人!

2. CyclicBarrier:同步屏障

  • CyclicBarrier 的字面解释是可循环使用的屏障,其功能为:让一组线程到达一个屏障(也可称为同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会打开,此时所有被屏障拦截的线程才会继续运行。
  • CyclicBarrier默认的构造器是CyclicBarrier(int parties),其参数表示屏障拦截的线程数量。每个线程调用await()方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞。
  • CyclicBarrier的另一个构造器public CyclicBarrier(int parties, Runnable barrierAction),用于在线程到达屏障时,优先执行barrierAction,方便处理复杂业务。
  • CyclicBarrier可以用于多线程计算数据,最后合并计算结果的场景(在barrierAction内部处理合并逻辑)。
package com.java.juc1;import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;/**
 * @author rrqstart
 * 案例:人到齐了再开会
 */publicclassCyclicBarrierTest{publicstaticvoidmain(String[] args){//public CyclicBarrier(int parties, Runnable barrierAction)
        CyclicBarrier cyclicBarrier=newCyclicBarrier(8,()->{
            System.out.println("7名员工已到齐,现在开会。");});for(int i=0; i<7; i++){newThread(()->{
                System.out.println(Thread.currentThread().getName()+"号员工已到会议室!");try{
                    cyclicBarrier.await();}catch(InterruptedException e){
                    e.printStackTrace();}catch(BrokenBarrierException e){
                    e.printStackTrace();}}, String.valueOf(i+1)).start();}try{
            cyclicBarrier.await();}catch(InterruptedException e){
            e.printStackTrace();}catch(BrokenBarrierException e){
            e.printStackTrace();}

        System.out.println("会议正在进行中......");}}
//执行结果:2号员工已到会议室!5号员工已到会议室!4号员工已到会议室!6号员工已到会议室!1号员工已到会议室!7号员工已到会议室!3号员工已到会议室!7名员工已到齐,现在开会。
会议正在进行中......

CountDownLatch 和 CyclicBarrier 的区别:

  • CountDownLatch 的计数器只能使用一次,CyclicBarrier 的计数器可以使用reset()方法重置,所以 CyclicBarrier 可以处理更复杂的业务,比如:如果计算发生错误,可以重置计数器,并让线程重新执行一次计算。
  • CyclicBarrier 还提供了其他有用的方法,比如:getNumberWaiting()方法可以获得 CyclicBarrier 阻塞的线程数量,isBroken()方法用来了解阻塞的线程是否被中断。

3. Semaphore:控制并发线程数

  • Semaphore(信号量)用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源。在Semaphore上定义了两种常用操作:
    • acquire(获取):当一个线程调用acquire操作时,它要么成功获取信号量(信号量减1),要么一直等下去,直到有其他线程释放信号量,或超时。
    • release(释放):实际上会将信号量的值加1,然后唤醒等待的线程。
  • 信号量主要用于两个目的:
    • ① 用于多个共享资源的互斥使用。
    • ② 用于并发线程数的控制(并发限流、流量控制)。
  • Semaphore 往往用于资源有限的场景中,去限制线程的数量。
package com.java.juc1;import java.util.concurrent.Semaphore;/**
 * @author rrqstart
 * Semaphore(信号量)相关案例:争车位 / 秒杀系统
 */publicclassSemaphoreTest{publicstaticvoidmain(String[] args){//Semaphore的构造器接收一个int型数据,表示可用的许可证数量。
        Semaphore semaphore=newSemaphore(3);//模拟资源类,有3个空车位//6个线程争夺3个资源,虽然有6个线程,但是只允许3个线程并发执行。for(int i=0; i<6; i++){newThread(()->{try{
                    semaphore.acquire();//获取许可证
                    System.out.println(Thread.currentThread().getName()+"号车已抢到车位!");
                    Thread.sleep(2000);//已抢到车位的车子停在车位上2秒
                    System.out.println(Thread.currentThread().getName()+"号车已离开车位!......");}catch(InterruptedException e){
                    e.printStackTrace();}finally{
                    semaphore.release();//归还许可证}}, String.valueOf(i+1)).start();}}}

执行结果:

1号车已抢到车位!3号车已抢到车位!2号车已抢到车位!1号车已离开车位!......3号车已离开车位!......4号车已抢到车位!2号车已离开车位!......5号车已抢到车位!6号车已抢到车位!5号车已离开车位!......6号车已离开车位!......4号车已离开车位!......

4. Exchanger:线程间交换数据

  • Exchanger(交换者)是一个用于线程间协作的工具类。Exchanger 用于进行线程间的数据交换。
  • 它提供一个同步点,在这个同步点,两个线程可以交换彼此的数据,这两个线程通过public V exchange(V x) throws InterruptedException方法交换数据,如果第一个线程先执行exchange(V x)方法,它会一直等待第二个线程也执行exchange(V x)方法,当两个线程都到达同步点时,这两个线程就可以交换数据。
  • 如果两个线程有一个没有执行exchange(V x)方法,则会一直等待。如果要避免一直等待,可以使用public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException方法设置最大等待时长。
  • Exchanger的应用场景:遗传算法、校对工作。
package com.java.juc1;import java.util.concurrent.Exchanger;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/**
 * @author rrqstart
 * @Description 银行流水校对工作
 */publicclassExchangerTest{privatestaticfinal Exchanger<String> EXCHANGER=newExchanger<>();privatestatic ExecutorService threadPool= Executors.newFixedThreadPool(2);publicstaticvoidmain(String[] args){
        threadPool.execute(()->{try{
                String a="银行流水a";//a员工录入的银行流水数据
                EXCHANGER.exchange(a);}catch(InterruptedException e){
                e.printStackTrace();}});

        threadPool.execute(()->{try{
                String b="银行流水b";//b员工录入的银行流水数据
                String a= EXCHANGER.exchange(b);
                System.out.println("a和b的数据是否一致:"+ a.equals(b));
                System.out.println("a录入的是:"+ a);
                System.out.println("b录入的是:"+ b);}catch(InterruptedException e){
                e.printStackTrace();}});
        threadPool.shutdown();}}
a和b的数据是否一致:false
a录入的是:银行流水a
b录入的是:银行流水b

5. Phaser:增强的CyclicBarrier

  • 作者:coder1qiang
  • 原文链接:https://blog.csdn.net/rrq_0324/article/details/110559812
    更新时间:2022-09-30 11:05:26