当前位置: 首页 > news >正文

JUC并发工具类

一、ReentrantLock

特点:独占、可重入、公平/非公平、可中断、支持多个条件变量

1、常用api

ReentrantLock实现了Lock接口,Lock类规范定义了如下方法

  • lock():获取锁,调用该方法的线程会获取锁,当锁获得后,该方法返回
  • lockInterruptibly():可中断得获取锁,和lock()方法不同之处在于该方法会响应中断,即在锁的获取中可以中断当前线程
  • tryLock():尝试非阻塞的获取锁,调用该方法后立即返回。如果能够获取到返回true,否则返回false
  • tryLock(long, TimeUnit):超时获取锁,当前线程在三种情况下会被返回(1、当前线程在超时时间内获取了锁 2、当前线程在超时时间内被中断 3、超时时间结束,返回false)
  • unLock():释放锁
  • newCondition():获取等待通知组件,该组件和当前的锁绑定,当前线程只有获取了锁,才能调用该组件的await()方法,而调用后,当前线程将释放锁

2、使用

使用范式:

 如果把lock.lock()加锁操作放在try里面,可能try里面其它代码导致加锁失败,最后lock.unlock()解锁操作时由于没加锁成功抛出IllegalMonitorStateException异常

public class ReentrantLockTest {private final ReentrantLock lock = new ReentrantLock();// 库存数量private static int inventoryQuantity = 5;// 减库存方法private void reduceInventory() {lock.lock();try {if (inventoryQuantity > 0) {try {Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"购买了商品,剩余库存数:"+--inventoryQuantity);} else {System.out.println(Thread.currentThread().getName()+":已经没有库存了");}} finally {lock.unlock();}}public static void main(String[] args) throws InterruptedException {ReentrantLockTest lockTest = new ReentrantLockTest();for (int i = 1; i <= 10; i++) {new Thread(() -> {lockTest.reduceInventory();}).start();}}
}

打印结果:

Thread-0购买了商品,剩余库存数:4
Thread-1购买了商品,剩余库存数:3
Thread-2购买了商品,剩余库存数:2
Thread-3购买了商品,剩余库存数:1
Thread-5购买了商品,剩余库存数:0
Thread-4:已经没有库存了
Thread-6:已经没有库存了
Thread-9:已经没有库存了
Thread-7:已经没有库存了
Thread-8:已经没有库存了

3、公平锁和非公平锁

ReentrantLock支持公平锁和非公平锁,默认是非公平锁

  • 公平锁:线程在获取锁时,按照等待的先后顺序获取锁
  • 非公平锁:线程在获取锁时,不按照等待的先后顺序获取锁,而是随机获取锁
ReentrantLock lock = new ReentrantLock(); //参数默认false,非公平锁
ReentrantLock lock = new ReentrantLock(true); //公平锁

非公平锁在加锁允许先进行CAS操作判断一次,公平锁则直接进入acquire()方法

4、可重入锁

可重入锁又名递归锁,是指在同一个线程在外层方法获取锁的时候,再进入该线程的内层方法会自动获取锁(前提锁对象得是同一个),不会因为之前已经获取过还没释放而阻塞。ReentrantLock和synchronized都是可重入锁,可重入锁可一定层度避免死锁。在实际开发中,可重入锁常常 应用于递归操作、调用同一个类中的其他方法、锁嵌套等场景中

public class ReentrantLockRecursiveTest {private final ReentrantLock lock = new ReentrantLock();/*** 递归调用5次* @param num*/public void recursiveCall(int num) {lock.lock();try {if (num > 5) {return;}System.out.println("执行递归调用第"+num+"次");recursiveCall(++num);} finally {lock.unlock();}}public static void main(String[] args) {ReentrantLockRecursiveTest lockRecursiveTest = new ReentrantLockRecursiveTest();lockRecursiveTest.recursiveCall(1);}
}

打印结果:

执行递归调用第1次
执行递归调用第2次
执行递归调用第3次
执行递归调用第4次
执行递归调用第5次

5、基于Condition的等待唤醒机制

java.util.concurrent类库中提供Condition类实现线程之间的协调。调用Condition.await()方法使线程等待,其它线程调用Condition.signal()或Condition.signalAll()方法唤醒等待的线程

注意:调用Condition的await()和signal()方法,都必须在lock保护之内

案例:基于ReentrantLock和Condition实现一个简单队列

public class ReentrantLockConditionTest {public static void main(String[] args) {Queue queue = new Queue(5);// 创建生产者线程new Thread(new Producer(queue)).start();// 创建消费者线程new Thread(new Customer(queue)).start();}}class Queue {private Object[] items;int size = 0;int takeIndex;int putIndex;private ReentrantLock lock;public Condition notEmpty;public Condition notFull;public Queue(int capacity) {this.items = new Object[capacity];lock = new ReentrantLock();notEmpty = lock.newCondition();notFull = lock.newCondition();}/*** 生产者方法* @param value* @throws InterruptedException*/public void put(Object value) throws InterruptedException {lock.lock();try {while (size == items.length) {// 队列满了 进入等待notFull.await();}items[putIndex] = value;if (++putIndex == items.length) {putIndex = 0;}size++;notEmpty.signal();// 队列中只要添加一个对象就唤醒消费者线程} finally {System.out.println("producer生产:"+value);lock.unlock();}}/*** 消费者方法* @return* @throws InterruptedException*/public Object take() throws InterruptedException {lock.lock();try {// 队列空了就让消费者等待while (size == 0) {notEmpty.await();}Object value = items[takeIndex];items[takeIndex] = null;if (++takeIndex == items.length) {takeIndex = 0;}size--;notFull.signal();// 队列中只要消费一个对象就唤醒生产者线程return value;} finally {lock.unlock();}}
}class Producer implements Runnable {private Queue queue;public Producer(Queue queue) {this.queue = queue;}@Overridepublic void run() {try {while (true) {Thread.sleep(1000);queue.put(new Random().nextInt(1000));}} catch (InterruptedException e) {e.printStackTrace();}}
}class Customer implements Runnable {private Queue queue;public Customer (Queue queue) {this.queue = queue;}@Overridepublic void run() {try {while (true) {Thread.sleep(2000);System.out.println("consumer消费:" + queue.take());}} catch (InterruptedException e) {e.printStackTrace();}}}

6、应用场景总结

ReentrantLock具体应用场景如下:

  1. 解决多线程竞争资源的问题,例如多个线程同时对同一个数据库进行写操作,可以使用ReentrantLock保证每次只有一个线程能够写入。
  2. 实现多线程任务的顺序执行,例如在一个线程执行完某个任务后,再让另一个线程执行任务。
  3. 实现多线程等待/通知机制,例如在某个线程执行完某个任务后,通知其他线程继续执行任务。

二、Semaphore

Semaphore(信号量)是一种用于多线程编程的同步工具,用于控制同时访问某个资源的线程数量。

Semaphore维护了一个计数器,线程可以通过调用acquire()方法来获取Semaphore中的许可证,当计数器为0时,调用acquire()的线程将被阻塞,直到有其他线程释放许可证;线程可以通过调用release()方法来释放Semaphore中的许可证,这会使Semaphore中的计数器增加,从而允许更多的线程访问共享资源。

1、常用api

  • permits 表示许可证的数量(资源数)
  • fair 表示公平性,如果这个设为 true 的话,下次执行的线程会是等待最久的线程
  • acquire() 表示阻塞并获取许可
  • tryAcquire() 方法在没有许可的情况下会立即返回 false,要获取许可的线程不会阻塞
  • release() 表示释放许可

2、使用

public class SemaphoreTest {// 定义两个资源数private static Semaphore semaphore = new Semaphore(2);private static Executor executor = Executors.newFixedThreadPool(10);public static void main(String[] args) {for (int i = 0; i < 10; i++) {executor.execute(() -> fluidControl());}}public static void fluidControl2() {try {// acquire()会构建同步等待队列semaphore.acquire();System.out.println("请求服务成功");Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();} finally {semaphore.release();}}public static void fluidControl() {// tryAcquire()直接CAS返回if (!semaphore.tryAcquire()) {System.out.println("请求被流控了");return;}try {System.out.println("请求服务成功");Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();} finally {semaphore.release();}}
}

3、应用场景

以下是一些使用Semaphore的常见场景:

  1. 限流:Semaphore可以用于限制对共享资源的并发访问数量,以控制系统的流量。
  2. 资源池:Semaphore可以用于实现资源池,以维护一组有限的共享资源。

三、CountDownLatch

CountDownLatch(闭锁)是一个同步协助类,允许一个或多个线程等待,直到其他线程完成操作集。

CountDownLatch使用给定的计数值(count)初始化。await方法会阻塞直到当前的计数值(count),由于countDown方法的调用达到0,count为0之后所有等待的线程都会被释放,并且随后对await方法的调用都会立即返回。这是一个一次性现象 —— count不会被重置。

1、常用api

  • CountDownLatch(int):构造方法初始化count数
  • await():等待count减到0后继续往后执行
  • await():等待指定时长,count值还没减到0,不再等待继续执行
  • countDown():每调用一次count就会减1,减到0为止 

2、使用

public class CountDownLatchTest {private static int[] values = {30, 20, 65, 23, 45};private static int result = 0;private static CountDownLatch coming = new CountDownLatch(values.length);public static void main(String[] args) throws InterruptedException {for (int i = 0; i < values.length; i++) {int tempI = i;new Thread(() -> {result += values[tempI];System.out.println(Thread.currentThread().getName()+"线程计算的结果集是:"+result);coming.countDown();}, "Thread_"+i).start();}coming.await();System.out.println("汇总结果集是:"+result);}
}

3、应用场景

以下是使用CountDownLatch的常见场景:

  1. 并行任务同步:CountDownLatch可以用于协调多个并行任务的完成情况,确保所有任务都完成后再继续执行下一步操作。
  2. 多任务汇总:CountDownLatch可以用于统计多个线程的完成情况,以确定所有线程都已完成工作。
  3. 资源初始化:CountDownLatch可以用于等待资源的初始化完成,以便在资源初始化完成后开始使用。

四、CyclicBarrier

CyclicBarrier(回环栅栏或循环屏障),是 Java 并发库中的一个同步工具,通过它可以实现让一组线程等待至某个状态(屏障点)之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后,CyclicBarrier可以被重用。

1、常用api

 // parties表示屏障拦截的线程数量,每个线程调用 await 方法告诉 CyclicBarrier 我已经到达了屏障,然后当前线程被阻塞。public CyclicBarrier(int parties)// 用于在线程到达屏障时,优先执行 barrierAction,方便处理更复杂的业务场景(该线程的执行时机是在到达屏障之后再执行)public CyclicBarrier(int parties, Runnable barrierAction)

//指定数量的线程全部调用await()方法时,这些线程不再阻塞
// BrokenBarrierException 表示栅栏已经被破坏,破坏的原因可能是其中一个线程 await() 时被中断或者超时
public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException//循环  通过reset()方法可以进行重置
public void reset()

2、使用

public class CyclicBarrierTest {public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(5);CyclicBarrier cyclicBarrier = new CyclicBarrier(5,() -> System.out.println("人齐了,准备发车"));for (int i = 0; i < 10; i++) {final int id = i+1;executorService.submit(new Runnable() {@Overridepublic void run() {try {System.out.println(id+"号马上就到");int sleepMills = ThreadLocalRandom.current().nextInt(2000);Thread.sleep(sleepMills);System.out.println(id + "号到了,上车");cyclicBarrier.await();} catch (InterruptedException e) {e.printStackTrace();}catch(BrokenBarrierException e){e.printStackTrace();}}});}}
}

3、应用场景

以下是一些常见的 CyclicBarrier 应用场景:

  1. 多线程任务:CyclicBarrier 可以用于将复杂的任务分配给多个线程执行,并在所有线程完成工作后触发后续操作。
  2. 数据处理:CyclicBarrier 可以用于协调多个线程间的数据处理,在所有线程处理完数据后触发后续操作。

4、CyclicBarrier 与 CountDownLatch 区别

  • CountDownLatch 是一次性的,CyclicBarrier 是可循环利用的
  • CountDownLatch 参与的线程的职责是不一样的,有的在倒计时,有的在等待倒计时结束。CyclicBarrier 参与的线程职责是一样的。

五、Exchanger

Exchanger是一个用于线程间协作的工具类,用于两个线程间交换数据。具体交换数据是通过exchange方法来实现的,如果一个线程先执行exchange方法,那么它会同步等待另一个线程也执行exchange方法,这个时候两个线程就都达到了同步点,两个线程就可以交换数据。

1、常用api

public V exchange(V x) throws InterruptedException
public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
  • V exchange(V v):等待另一个线程到达此交换点(除非当前线程被中断),然后将给定的对象传送给该线程,并接收该线程的对象。
  • V exchange(V v, long timeout, TimeUnit unit):等待另一个线程到达此交换点,或者当前线程被中断——抛出中断异常;又或者是等候超时——抛出超时异常,然后将给定的对象传送给该线程,并接收该线程的对象。

2、使用

public class ExchangerTest {private static Exchanger exchanger = new Exchanger();static String goods = "电脑";static String money = "$4000";public static void main(String[] args) throws InterruptedException {System.out.println("准备交易,一手交钱一手交货...");// 卖家new Thread(new Runnable() {@Overridepublic void run() {System.out.println("卖家到了,已经准备好货:" + goods);try {String money = (String) exchanger.exchange(goods);System.out.println("卖家收到钱:" + money);} catch (Exception e) {e.printStackTrace();}}}).start();Thread.sleep(3000);// 买家new Thread(new Runnable() {@Overridepublic void run() {try {System.out.println("买家到了,已经准备好钱:" + money);String goods = (String) exchanger.exchange(money);System.out.println("买家收到货:" + goods);} catch (Exception e) {e.printStackTrace();}}}).start();}
}

3、应用场景

Exchanger 可以用于各种应用场景,具体取决于具体的 Exchanger 实现。常见的场景包括:

  1. 数据交换:在多线程环境中,两个线程可以通过 Exchanger 进行数据交换。
  2. 数据采集:在数据采集系统中,可以使用 Exchanger 在采集线程和处理线程间进行数据交换。

六、Phaser

Phaser(阶段协同器)是一个Java实现的并发工具类,用于协调多个线程的执行。它提供了一些方便的方法来管理多个阶段的执行,可以让程序员灵活地控制线程的执行顺序和阶段性的执行。Phaser可以被视为CyclicBarrier和CountDownLatch的进化版,它能够自适应地调整并发线程数,可以动态地增加或减少参与线程的数量。所以Phaser特别适合使用在重复执行或者重用的情况。

1、常用api

构造方法

  • Phaser(): 参与任务数0
  • Phaser(int parties) :指定初始参与任务数
  • Phaser(Phaser parent) :指定parent阶段器, 子对象作为一个整体加入parent对象, 当子对象中没有参与者时,会自动从parent对象解除注册
  • Phaser(Phaser parent,int parties) : 集合上面两个方法

增减参与任务数方法

  • int register() 增加一个任务数,返回当前阶段号。
  • int bulkRegister(int parties) 增加指定任务个数,返回当前阶段号。
  • int arriveAndDeregister() 减少一个任务数,返回当前阶段号。

到达、等待方法

  • int arrive() 到达(任务完成),返回当前阶段号。
  • int arriveAndAwaitAdvance() 到达后等待其他任务到达,返回到达阶段号。
  • int awaitAdvance(int phase) 在指定阶段等待(必须是当前阶段才有效)
  • int awaitAdvanceInterruptibly(int phase) 阶段到达触发动作
  • int awaitAdvanceInterruptiBly(int phase,long timeout,TimeUnit unit)
  • protected boolean onAdvance(int phase,int registeredParties)类似CyclicBarrier的触发命令,通过重写该方法来增加阶段到达动作,该方法返回true将终结Phaser对象。

2、使用

public class PhaserBatchProcessorTest {public static void main(String[] args) {final Phaser phaser = new Phaser() {//重写该方法来增加阶段到达动作@Overrideprotected boolean onAdvance(int phase, int registeredParties) {// 参与者数量,去除主线程int staffs = registeredParties - 1;switch (phase) {case 0:System.out.println("大家都到公司了,出发去公园,人数:" + staffs);break;case 1:System.out.println("大家都到公园门口了,出发去餐厅,人数:" + staffs);break;case 2:System.out.println("大家都到餐厅了,开始用餐,人数:" + staffs);break;}// 判断是否只剩下主线程(一个参与者),如果是,则返回true,代表终止return registeredParties == 1;}};// 注册主线程 ———— 让主线程全程参与phaser.register();final StaffTask staffTask = new StaffTask();// 3个全程参与团建的员工for (int i = 0; i < 3; i++) {// 添加任务数phaser.register();new Thread(() -> {try {staffTask.step1Task();//到达后等待其他任务到达phaser.arriveAndAwaitAdvance();staffTask.step2Task();phaser.arriveAndAwaitAdvance();staffTask.step3Task();phaser.arriveAndAwaitAdvance();staffTask.step4Task();// 完成了,注销离开phaser.arriveAndDeregister();} catch (InterruptedException e) {e.printStackTrace();}}).start();}// 两个不聚餐的员工加入for (int i = 0; i < 2; i++) {phaser.register();new Thread(() -> {try {staffTask.step1Task();phaser.arriveAndAwaitAdvance();staffTask.step2Task();System.out.println("员工【" + Thread.currentThread().getName() + "】回家了");// 完成了,注销离开phaser.arriveAndDeregister();} catch (InterruptedException e) {e.printStackTrace();}}).start();}while (!phaser.isTerminated()) {int phase = phaser.arriveAndAwaitAdvance();if (phase == 2) {// 到了去餐厅的阶段,又新增4人,参加晚上的聚餐for (int i = 0; i < 4; i++) {phaser.register();new Thread(() -> {try {staffTask.step3Task();phaser.arriveAndAwaitAdvance();staffTask.step4Task();// 完成了,注销离开phaser.arriveAndDeregister();} catch (InterruptedException e) {e.printStackTrace();}}).start();}}}}static final Random random = new Random();static class StaffTask {public void step1Task() throws InterruptedException {// 第一阶段:来公司集合String staff = "员工【" + Thread.currentThread().getName() + "】";System.out.println(staff + "从家出发了……");Thread.sleep(random.nextInt(5000));System.out.println(staff + "到达公司");}public void step2Task() throws InterruptedException {// 第二阶段:出发去公园String staff = "员工【" + Thread.currentThread().getName() + "】";System.out.println(staff + "出发去公园玩");Thread.sleep(random.nextInt(5000));System.out.println(staff + "到达公园门口集合");}public void step3Task() throws InterruptedException {// 第三阶段:去餐厅String staff = "员工【" + Thread.currentThread().getName() + "】";System.out.println(staff + "出发去餐厅");Thread.sleep(random.nextInt(5000));System.out.println(staff + "到达餐厅");}public void step4Task() throws InterruptedException {// 第四阶段:就餐String staff = "员工【" + Thread.currentThread().getName() + "】";System.out.println(staff + "开始用餐");Thread.sleep(random.nextInt(5000));System.out.println(staff + "用餐结束,回家");}}
}

3、应用场景

以下是一些常见的 Phaser 应用场景:

  1. 多线程任务分配:Phaser 可以用于将复杂的任务分配给多个线程执行,并协调线程间的合作。
  2. 多级任务流程:Phaser 可以用于实现多级任务流程,在每一级任务完成后触发下一级任务的开始。
  3. 模拟并行计算:Phaser 可以用于模拟并行计算,协调多个线程间的工作。
  4. 阶段性任务:Phaser 可以用于实现阶段性任务,在每一阶段任务完成后触发下一阶段任务的开始。
http://www.lryc.cn/news/99276.html

相关文章:

  • 【雕爷学编程】MicroPython动手做(10)——零基础学MaixPy之神经网络KPU
  • MySQL~SQL语句
  • 从零开始构建基于YOLOv5的目标检测系统
  • PDF尺寸修改:等比绽放(标准面单100*150mm)
  • C++ - list介绍 和 list的模拟实现
  • 干翻Dubbo系列第四篇:Dubbo3第一个应用程序细节补充
  • 深度学习初探
  • nn.BCELoss与nn.CrossEntropyLoss
  • CSDN浏览如何解决
  • web前端开发小知识
  • Java泛型的简单认识
  • 视频转化为图片或灰度视频
  • 【动态规划刷题 2】使⽤最⼩花费爬楼梯 解码⽅法
  • Python的基本语法
  • Kubernetes那点事儿——存储之存储卷
  • Go语言中‘String’包中的‘Cut‘函数的实现
  • 【JAVASE】顺序和选择结构
  • Oracle恢复删除的数据
  • (无人机方向)ros小白之键盘控制无人机(终端方式)
  • 【python学习笔记】argparse --- 命令行选项、参数和子命令解析器
  • 【Java框架】RPC远程调用
  • 云原生全栈体系(一)
  • 【【51单片机直流电机调速】】
  • 【Spring Boot】
  • 使用docker 部署自己的chatgpt
  • Python适配器模式介绍、使用方法
  • 【数据结构】复杂度
  • 【读点论文】PP-YOLOE: An evolved version of YOLO,面向友好部署的模型设计,为项目后续产业落地提供了更加有效的参考
  • 微服务入门---SpringCloud(二)
  • 51单片机IO口控制