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