多线程系列(十五) -常用并发工具类详解

一、择要

在前几篇文章中，我们讲到了线程、线程池、BlockingQueue 等焦点组件，其实 JDK 给开发者还提供了比synchronized更加高级的线程同步组件，好比 CountDownLatch、CyclicBarrier、Semaphore、Exchanger 等并发工具类。
下面我们一起来相识一下这些常用的并发工具类！
二、常用并发工具类

2.1、CountDownLatch

CountDownLatch是 JDK5 之后参加的一种并发流程控制工具类，它允许一个或多个线程一直等待，直到其他线程运行完成后再执行。
它的工作原理重要是通过一个计数器来实现，初始化的时候必要指定线程的数量；每当一个线程完成了自己的任务，计数器的值就相应得减 1；当计数器到达 0 时，表示所有的线程都已经执行完毕，处于等待的线程就可以恢复继续执行任务。
根据CountDownLatch的工作原理，它的应用场景一样平常可以分别为两种：

• 场景一：某个线程必要在其他 n 个线程执行完毕后，再继续执行
  
• 场景二：多个工作线程等待某个线程的下令，同时执行同一个任务
  

下面我们先来看下两个简朴的示例。
示例1：某个线程等待 n 个工作线程

好比某项任务，先接纳多线程去执行，最后必要在主线程中进行汇总处理，这个时候CountDownLatch就可以发挥作用了，详细应用如下！

1. public class CountDownLatchTest {
3.     public static void main(String[] args) throws InterruptedException {
4.         // 采用 10 个工作线程去执行任务
5.         final int threadCount = 10;
6.         CountDownLatch countDownLatch = new CountDownLatch(threadCount);
7.         for (int i = 0; i < threadCount; i++) {
8.             new Thread(new Runnable() {
9.                 @Override
10.                 public void run() {
11.                     // 执行具体任务
12.                     System.out.println("thread name:" +  Thread.currentThread().getName() + "，执行完毕！");
13.                     // 计数器减 1
14.                     countDownLatch.countDown();
15.                 }
16.             }).start();
17.         }
19.         // 阻塞等待 10 个工作线程执行完毕
20.         countDownLatch.await();
21.         System.out.println("所有任务线程已执行完毕，准备进行结果汇总");
22.     }
23. }

复制代码

运行结果如下：

1. thread name:Thread-0，执行完毕！
2. thread name:Thread-2，执行完毕！
3. thread name:Thread-1，执行完毕！
4. thread name:Thread-3，执行完毕！
5. thread name:Thread-4，执行完毕！
6. thread name:Thread-5，执行完毕！
7. thread name:Thread-6，执行完毕！
8. thread name:Thread-7，执行完毕！
9. thread name:Thread-8，执行完毕！
10. thread name:Thread-9，执行完毕！
11. 所有任务线程执行完毕，准备进行结果汇总

复制代码

示例2：n 个工作线程等待某个线程

好比田径赛跑，10 个同砚预备开跑，但是必要等工作人员发出枪声才允许开跑，利用CountDownLatch可以实现这一功能，详细应用如下！

1. public class CountDownLatchTest {
3.     public static void main(String[] args) throws InterruptedException {
4.         // 使用一个计数器
5.         CountDownLatch countDownLatch = new CountDownLatch(1);
6.         final int threadCount = 10;
7.         // 采用 10 个工作线程去执行任务
8.         for (int i = 0; i < threadCount; i++) {
9.             new Thread(new Runnable() {
10.                 @Override
11.                 public void run() {
12.                     try {
13.                         // 阻塞等待计数器为 0
14.                         countDownLatch.await();
15.                     } catch (InterruptedException e) {
16.                         e.printStackTrace();
17.                     }
18.                     // 发起某个服务请求，省略
19.                     System.out.println("thread name:" +  Thread.currentThread().getName() + "，开始执行！");
21.                 }
22.             }).start();
23.         }
25.         Thread.sleep(1000);
26.         System.out.println("thread name:" +  Thread.currentThread().getName() + " 准备开始！");
27.         // 将计数器减 1，运行完成后为 0
28.         countDownLatch.countDown();
29.     }
30. }

复制代码

运行结果如下：

1. thread name:main 准备开始！
2. thread name:Thread-0，开始执行！
3. thread name:Thread-1，开始执行！
4. thread name:Thread-2，开始执行！
5. thread name:Thread-3，开始执行！
6. thread name:Thread-5，开始执行！
7. thread name:Thread-6，开始执行！
8. thread name:Thread-8，开始执行！
9. thread name:Thread-7，开始执行！
10. thread name:Thread-4，开始执行！
11. thread name:Thread-9，开始执行！

复制代码

从上面的示例可以很清晰的看到，CountDownLatch类似于一个倒计数器，当计数器为 0 的时候，调用await()方法的线程会被解除等待状态，然后继续执行。
CountDownLatch类的重要方法，有以下几个：

• public CountDownLatch(int count)：焦点构造方法，初始化的时候必要指定线程数
  
• countDown()：每调用一次，计数器值 -1，直到 count 被减为 0，表示所有线程全部执行完毕
  
• await()：等待计数器变为 0，即等待所有异步线程执行完毕，否则一直阻塞
  
• await(long timeout, TimeUnit unit)：支持指定时间内的等待，避免永久阻塞，await()的一个重载方法
  

从以上的分析可以得出，当计数器为 1 的时候，即由一个线程来关照其他线程，结果等同于对象的wait()和notifyAll()；当计时器大于 1 的时候，可以实现多个工作线程完成任务后关照一个大概多个等待线程继续工作，CountDownLatch可以当作是一种进阶版的等待/关照机制，在实际中应用比力多见。
2.2、CyclicBarrier

CyclicBarrier从字面上很容易理解，表示可循环利用的屏障，它真正的作用是让一组线程到达一个屏障时被阻塞，直到满足要求的线程数都到达屏障时，屏障才会解除，此时所有被屏障阻塞的线程就可以继续执行。
下面我们还是先看一个简朴的示例，以便于更好的理解这个工具类。

1. public class CyclicBarrierTest {
3.     public static void main(String[] args) {
4.         // 设定参与线程的个数为 5
5.         int threadCount = 5;
6.         CyclicBarrier cyclicBarrier = new CyclicBarrier(threadCount, new Runnable() {
7.             @Override
8.             public void run() {
9.                 System.out.println("所有的线程都已经准备就绪...");
10.             }
11.         });
12.         for (int i = 0; i < threadCount; i++) {
13.             new Thread(new Runnable() {
14.                 @Override
15.                 public void run() {
16.                     System.out.println("thread name:" +  Thread.currentThread().getName() + "，已达到屏障！");
17.                     try {
18.                         cyclicBarrier.await();
19.                     } catch (Exception e) {
20.                         e.printStackTrace();
21.                     }
22.                     System.out.println("thread name:" +  Thread.currentThread().getName() + "，阻塞解除，继续执行！");
23.                 }
24.             }).start();
25.         }
26.     }
27. }

复制代码

输出结果：

1. thread name:Thread-0，已达到屏障！
2. thread name:Thread-1，已达到屏障！
3. thread name:Thread-2，已达到屏障！
4. thread name:Thread-3，已达到屏障！
5. thread name:Thread-4，已达到屏障！
6. 所有的线程都已经准备就绪...
7. thread name:Thread-4，阻塞解除，继续执行！
8. thread name:Thread-0，阻塞解除，继续执行！
9. thread name:Thread-3，阻塞解除，继续执行！
10. thread name:Thread-1，阻塞解除，继续执行！
11. thread name:Thread-2，阻塞解除，继续执行！

复制代码

从上面的示例可以很清晰的看到，CyclicBarrier中设定的线程数相称于一个屏障，当所有的线程数达到时，此时屏障就会解除，线程继续执行剩下的逻辑。
CyclicBarrier类的重要方法，有以下几个：

• public CyclicBarrier(int parties)：构造方法，parties参数表示参与线程的个数
  
• public CyclicBarrier(int parties, Runnable barrierAction)：焦点构造方法，barrierAction参数表示线程到达屏障时的回调方法
  
• public void await()：焦点方法，每个线程调用await()方法告诉CyclicBarrier我已经到达了屏障，然后当火线程被阻塞，直到屏障解除，继续执行剩下的逻辑
  

从以上的示例中，可以看到CyclicBarrier与CountDownLatch有很多的相似之处，都能够实现线程之间的等待，但是它们的侧重点差异：

• CountDownLatch一样平常用于一个或多个线程，等待其他的线程执行完任务后再执行
  
• CyclicBarrier一样平常用于一组线程等待至某个状态，当状态解除之后，这一组线程再继续执行
  
• CyclicBarrier中的计数器可以反复利用，而CountDownLatch用完之后只能重新初始化
  

2.3、Semaphore

Semaphore通常我们把它称之为信号计数器，它可以保证同一时候最多有 N 个线程能访问某个资源，好比同一时候最多允许 10 个用户访问某个服务，同一时候最多创建 100 个数据库毗连等等。
Semaphore可以用于控制并发的线程数，实际应用场景非常的广，好比流量控制、服务限流等等。
下面我们看一个简朴的示例。

1. public class SemaphoreTest {
3.     public static void main(String[] args) {
4.         SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
6.         // 同一时刻仅允许最多3个线程获取许可
7.         final Semaphore semaphore = new Semaphore(3);
8.         // 初始化 5 个线程生成
9.         for (int i = 0; i < 5; i++) {
10.             new Thread(new Runnable() {
11.                 @Override
12.                 public void run() {
13.                     try {
14.                         // 如果超过了许可数量,其他线程将在此等待
15.                         semaphore.acquire();
16.                         System.out.println(format.format(new Date()) +  " thread name:" +  Thread.currentThread().getName() + " 获取许可，开始执行任务");
17.                         // 假设执行某项任务的耗时
18.                         Thread.sleep(2000);
19.                     } catch (Exception e) {
20.                         e.printStackTrace();
21.                     } finally {
22.                         // 使用完后释放许可
23.                         semaphore.release();
24.                     }
25.                 }
26.             }).start();
27.         }
28.     }
29. }

复制代码

输出结果：

1. 2023-11-22 17:32:01 thread name:Thread-0 获取许可，开始执行任务
2. 2023-11-22 17:32:01 thread name:Thread-1 获取许可，开始执行任务
3. 2023-11-22 17:32:01 thread name:Thread-2 获取许可，开始执行任务
4. 2023-11-22 17:32:03 thread name:Thread-4 获取许可，开始执行任务
5. 2023-11-22 17:32:03 thread name:Thread-3 获取许可，开始执行任务

复制代码

从上面的示例可以很清晰的看到，同一时候前 3 个线程获得了许可优先执行， 2 秒事后许可被开释，剩下的 2 个线程获取开释的许可继续执行。
Semaphore类的重要方法，有以下几个：

• public Semaphore(int permits)：构造方法，permits参数表示同一时间能访问某个资源的线程数量
  
• acquire()：获取一个许可，在获取到许可之前大概被其他线程调用停止之前，线程将一直处于阻塞状态
  
• tryAcquire(long timeout, TimeUnit unit)：表示在指定时间内实验获取一个许可，如果获取成功，返回true；反之false
  
• release()：开释一个许可，同时唤醒一个获取许可不成功的阻塞线程。
  

通过permits参数的设定，可以实现限制多个线程同时访问服务的结果，当permits参数为 1 的时候，表示同一时候只有一个线程能访问服务，相称于一个互斥锁，结果等同于synchronized。
利用Semaphore的时候，通常必要先调用acquire()大概tryAcquire()获取许可，然后通过try ... finally模块在finally中开释许可。
例如如下方式，实验在 3 秒内获取许可，如果没有获取就退出，防止程序一直阻塞。

1. // 尝试 3 秒内获取许可
2. if(semaphore.tryAcquire(3, TimeUnit.SECONDS)){
3.     try {
4.        // ...业务逻辑
5.     }  finally {
6.         // 释放许可
7.         semaphore.release();
8.     }
9. }

复制代码

2.4、Exchanger

Exchanger从字面上很容易理解表示交换，它重要用途在两个线程之间进行数据交换，注意也只能在两个线程之间进行数据交换。
Exchanger提供了一个exchange()同步交换方法，当两个线程调用exchange()方法时，无论调用时间先后，会相互等待线程到达exchange()方法同步点，此时两个线程进行交换数据，将本线程产出数据传递给对方。
简朴的示例如下。

1. public class ExchangerTest {
3.     public static void main(String[] args) {
4.         // 交换同步器
5.         Exchanger<String> exchanger = new Exchanger<>();
7.         // 线程1
8.         new Thread(new Runnable() {
9.             @Override
10.             public void run() {
11.                 try {
12.                     String value = "A";
13.                     System.out.println("thread name:" +  Thread.currentThread().getName() + " 原数据：" + value);
14.                     String newValue = exchanger.exchange(value);
15.                     System.out.println("thread name:" +  Thread.currentThread().getName() + " 交换后的数据：" + newValue);
16.                 } catch (InterruptedException e) {
17.                     e.printStackTrace();
18.                 }
19.             }
20.         }).start();
22.         // 线程2
23.         new Thread(new Runnable() {
24.             @Override
25.             public void run() {
26.                 try {
27.                     String value = "B";
28.                     System.out.println("thread name:" +  Thread.currentThread().getName() + " 原数据：" + value);
29.                     String newValue = exchanger.exchange(value);
30.                     System.out.println("thread name:" +  Thread.currentThread().getName() + " 交换后的数据：" + newValue);
31.                 } catch (InterruptedException e) {
32.                     e.printStackTrace();
33.                 }
34.             }
35.         }).start();
36.     }
37. }

复制代码

输出结果：

1. thread name:Thread-0 原数据：A
2. thread name:Thread-1 原数据：B
3. thread name:Thread-0 交换后的数据：B
4. thread name:Thread-1 交换后的数据：A

复制代码

从上面的示例可以很清晰的看到，当线程Thread-0和Thread-1都到达了exchange()方法的同步点时，进行了数据交换。
Exchanger类的重要方法，有以下几个：

• exchange(V x)：等待另一个线程到达此交换点，然后将给定的对象传送给该线程，并接收该线程的对象，除非当火线程被停止，否则一直阻塞等待
  
• exchange(V x, long timeout, TimeUnit unit)：表示在指定的时间内等待另一个线程到达此交换点，如果超时会自动退出并抛超时异常
  

如果多个线程调用exchange()方法，数据交换可能会出现混乱，因此实际上Exchanger应用并不多见。
三、小结

本文重要围绕 Java 多线程中常见的并发工具类进行了简朴的用例介绍，这些工具类都可以实现线程同步的结果，底层原理实现重要是基于 AQS 队列式同步器来实现，关于 AQS 我们会在后期的文章中再次介绍。
本文篇幅稍有所长，内容难免有所遗漏，接待大家留言指出！
四、参考

1.https://www.cnblogs.com/xrq730/p/4869671.html
2.https://zhuanlan.zhihu.com/p/97055716

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。