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 可以用于实现阶段性任务，在每一阶段任务完成后触发下一阶段任务的开始。