Java 中的并发包指的是 java.util.concurrent(简称 JUC)包和其子包下的类和接口,它为 Java 的并发提供了各种功能支持,比如:
- 提供了线程池的创建类 ThreadPoolExecutor、Executors 等;
- 提供了各种锁,如 Lock、ReentrantLock 等;
- 提供了各种线程安全的数据结构,如 ConcurrentHashMap、LinkedBlockingQueue、DelayQueue 等;
- 提供了更加高级的线程同步结构,如 CountDownLatch、CyclicBarrier、Semaphore 等。
在前面的章节中我们已经详细地介绍了线程池的使用、线程安全的数据结构等,本文我们就重点学习一下 Java 并发包中更高级的线程同步类:CountDownLatch、CyclicBarrier、Semaphore 和 Phaser 等。
CountDownLatch 介绍和使用
CountDownLatch(闭锁)可以看作一个只能做减法的计数器,可以让一个或多个线程等待执行。
CountDownLatch 有两个重要的方法:
- countDown():使计数器减 1;
- await():当计数器不为 0 时,则调用该方法的线程阻塞,当计数器为 0 时,可以唤醒等待的一个或者全部线程。
CountDownLatch 使用场景:
以生活中的情景为例,比如去医院体检,通常人们会提前去医院排队,但只有等到医生开始上班,才能正式开始体检,医生也要给所有人体检完才能下班,这种情况就要使用 CountDownLatch,流程为:患者排队 → 医生上班 → 体检完成 → 医生下班。
CountDownLatch 示例代码如下:
// 医院闭锁
CountDownLatch hospitalLatch = new CountDownLatch(1);
// 患者闭锁
CountDownLatch patientLatch = new CountDownLatch(5);
System.out.println("患者排队");
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
final int j = i;
executorService.execute(() -> {
try {
hospitalLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("体检:" + j);
patientLatch.countDown();
});
}
System.out.println("医生上班");
hospitalLatch.countDown();
patientLatch.await();
System.out.println("医生下班");
executorService.shutdown();
以上程序执行结果如下:
> 患者排队
>
> 医生上班
>
> 体检:4
>
> 体检:0
>
> 体检:1
>
> 体检:3
>
> 体检:2
>
> 医生下班
执行流程如下图:
imageCyclicBarrier 介绍和使用
CyclicBarrier(循环屏障)通过它可以实现让一组线程等待满足某个条件后同时执行。
CyclicBarrier 经典使用场景是公交发车,为了简化理解我们这里定义,每辆公交车只要上满 4 个人就发车,后面来的人都会排队依次遵循相应的标准。
它的构造方法为 CyclicBarrier(int parties,Runnable barrierAction)
其中,parties 表示有几个线程来参与等待,barrierAction 表示满足条件之后触发的方法。CyclicBarrier 使用 await() 方法来标识当前线程已到达屏障点,然后被阻塞。
CyclicBarrier 示例代码如下:
import java.util.concurrent.*;
public class CyclicBarrierTest {
public static void main(String[] args) throws InterruptedException {
CyclicBarrier cyclicBarrier = new CyclicBarrier(4, new Runnable() {
@Override
public void run() {
System.out.println("发车了");
}
});
for (int i = 0; i < 4; i++) {
new Thread(new CyclicWorker(cyclicBarrier)).start();
}
}
static class CyclicWorker implements Runnable {
private CyclicBarrier cyclicBarrier;
CyclicWorker(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run() {
for (int i = 0; i < 2; i++) {
System.out.println("乘客:" + i);
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}
}
}
以上程序执行结果如下:
> 乘客:0
>
> 乘客:0
>
> 乘客:0
>
> 乘客:0
>
> 发车了
>
> 乘客:1
>
> 乘客:1
>
> 乘客:1
>
> 乘客:1
>
> 发车了
执行流程如下图:
imageSemaphore 介绍和使用
Semaphore(信号量)用于管理多线程中控制资源的访问与使用。Semaphore 就好比停车场的门卫,可以控制车位的使用资源。比如来了 5 辆车,只有 2 个车位,门卫可以先放两辆车进去,等有车出来之后,再让后面的车进入。
Semaphore 示例代码如下:
Semaphore semaphore = new Semaphore(2);
ThreadPoolExecutor semaphoreThread = new ThreadPoolExecutor(10, 50, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
for (int i = 0; i < 5; i++) {
semaphoreThread.execute(() -> {
try {
// 堵塞获取许可
semaphore.acquire();
System.out.println("Thread:" + Thread.currentThread().getName() + " 时间:" + LocalDateTime.now());
TimeUnit.SECONDS.sleep(2);
// 释放许可
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
以上程序执行结果如下:
> Thread:pool-1-thread-1 时间:2020-12-10 21:18:42
>
> Thread:pool-1-thread-2 时间:2020-12-10 21:18:42
>
> Thread:pool-1-thread-3 时间:2020-12-10 21:18:44
>
> Thread:pool-1-thread-4 时间:2020-12-10 21:18:44
>
> Thread:pool-1-thread-5 时间:2020-12-10 21:18:46
执行流程如下图:
enter image description here
网友评论