在上一篇博客中我们深入剖析了 J.U.C 的底层灵魂AQSAbstractQueuedSynchronizer。理解了 AQS 的state状态控制与双向阻塞队列之后再来看 J.U.C 提供的各种高级并发工具类就会发现它们不过是 AQS 在不同业务场景下的“优雅包装”。在实际开发中除了传统的显式锁Lock我们还经常遇到以下几类典型的多线程协作需求主线程等待多个子线程全部完成后再继续如火箭发射前的各项指标检查。多个子线程互相等待到齐后再同时冲锋如旅行团人齐了才发车且可以重复利用。限制同时访问某资源的线程数量如秒杀系统的接口限流、数据库连接池。两个线程安全地对调/交换数据如双缓冲区交换、遗传算法中的交叉配对。针对这四大需求J.U.C 提供了非常经典的并发工具“四大神兽”CountDownLatch、CyclicBarrier、Semaphore和Exchanger。一.等待多线程的CountDownLatchCountDownLatch倒计时器允许一个或多个线程等待其他线程完成一组操作。它的核心思想是初始化一个计数器N每次有线程完成任务就将计数器减 1当计数器递减到 0 时所有在await()上等待的线程才会被唤醒并继续执行。核心方法new CountDownLatch(int count)初始化计数器。countDown()计数器减 1非阻塞任何线程都可以调用。await()当前线程挂起阻塞直到计数器变为 0。await(long timeout, TimeUnit unit)带超时的等待。初始CountDownLatchCountDownLatch同步工具允许一条或多条线程等待其他线程中的一组操作后再继续执行比如我需要有七颗龙珠才能召唤神龙这七颗龙珠互相没有关系如果我一颗一颗收集太慢了所以适合派七个人每个人帮我收集一颗对比下来就是:子线程:各自寻找相应的目标龙珠主线程:等子线程都执行完毕了再召唤神龙。CountDownLatch验证public class Main { private final static Random randomnew Random(); static class SearchTask implements Runnable{ private Integer id; private CountDownLatch latch; public SearchTask(Integer id, CountDownLatch latch) { this.id id; this.latch latch; } Override public void run() { System.out.println(开始寻找id号龙珠); int secondsrandom.nextInt(10); try { Thread.sleep(seconds*1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(找到id号龙珠耗时seconds秒); latch.countDown();//告知任务执行完毕 } } public static void main(String[] args) { ListInteger idList Arrays.asList(1,2,3,4,5,6,7); CountDownLatch latchnew CountDownLatch(idList.size()); for (Integer id:idList){ new Thread(new SearchTask(id,latch)).start(); } try { latch.await();//等待所有线程执行完毕 } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(所有龙珠都找到啦); } }其中涉及到的方法:public void await() throws InterruptedException //阻塞当前线程直到计数器值变为0。如果计数器已经为0立即返回。如果在等待过程中被中断抛出 InterruptedException 异常。 public void countDown() //递减计数器的值表示一个任务已经完成。执行结果:二.同步屏障 CyclicBarrierCyclicBarrier的字面意思是可以循环使用(Cycle)的屏障(Barrier).要做的事情是让一组线程到达一个屏障(也可以叫同步点)时被阻塞直到最后一个线程到达屏障时屏障才会开门所有被屏障拦截的线程才会继续运行类比理解:坐过那种滚动发车的汽车吗例如汽车作为10个坐满就走那前9个坐上车的人就只有等待第10个人来了之后就发车一起出发。代码示例:11个空位主线程和子线程看成一样的不用区别理解。前10个线程到达之后只能等待等到第11个线程到来之后全部一起执行。public class Main { public static void main(String[] args) throws InterruptedException, BrokenBarrierException { CyclicBarrier cbnew CyclicBarrier(11); for (int i 1; i 10 ; i) { String name线程i; new Thread(new Runnable() { Override public void run() { System.out.println(任务:Thread.currentThread().getName()正在等待....); try { cb.await(); } catch (InterruptedException | BrokenBarrierException e) { throw new RuntimeException(e); } System.out.println(任务:Thread.currentThread().getName()完成执行); } }, name).start(); } Thread.sleep(200); System.out.println(任务11还要5s才到其它任务先等待...); Thread.sleep(5000); cb.await(); System.out.println(任务11开始执行...); } }CountDownLatch的计数器只能使用一次而CyclicBarrier的计数器可以使用reset()方法重置。所以CyclicBarrier能处理更为复杂的业务场景。例如如果计算发生错误可以重置计数器并让线程重新执行一次。CyclicBarrier还提供其他有用的方法比如getNumberWaiting方法可以获得Cyclic-Barrier 阻塞的线程数量等。三.控制并发线程数的SemaphoreSemaphore(信号量)是用来控制同时访问特定资源的线程数量它通过协调各个线程以保证合理的使用公共资源。可以把它想象成一个拥有固定停车位的停车场停车场总共有 N个车位许可证permits。入场车辆调用acquire()拿走一张通行证如果车位满了后面的车必须在入口排队等待。离场车辆调用release()归还一张通行证入口排队的车辆才能拿到通行证驶入。核心方法new Semaphore(int permits)创建指定许可数量的信号量默认非公平锁。new Semaphore(int permits, boolean fair)支持创建公平/非公平信号量。acquire()获取一个许可若无可用许可则阻塞支持响应中断。tryAcquire()尝试获取许可拿不到立马返回false不阻塞。release()释放一个许可归还给信号量。public class Main { public static void main(String[] args) throws InterruptedException { Semaphore semaphorenew Semaphore(5); for(int i0;i10;i){ new Thread(new Runnable() { Override public void run() { try { semaphore.acquire(); System.out.println(Thread.currentThread().getName()正在使用); Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); }finally { semaphore.release(); } } }).start(); } Thread.sleep(20000); } }虽然会有20个线程尝试执行但是同一时间只会有5个线程可以运行构造方法 Semaphore(int permmits)接受一个整型的数字表示可用许可证数量。Semaphore(5)表示允许5个线程获取许可证也就是最大并发数是5。Semaphore的用法也很简单首先线程使用 Semaphore的acquire()方法获取一个许可证使用完之后调用release()方法归还许可证。还可以用tryAcquire()方法尝试获取许可证。四.线程间交换数据的ExchangerExchanger(交换者)是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据如果第一个线程先执行exchange()方法它会一直等待第二个线程也 执行exchange方法当两个线程都到达同步点时这两个线程就可以交换数据将本线程生产出来的数据传递给对方。exchange()方法:入参是本线程需要传递给对方的数据返回值则是对方线程传过来的数据。如果两个线程有一个没有执行exchange()方法则会一直等待如果担心有特殊情况发生避免一直等待可以使用exchange(VxlongtimeoutTimeUnit unit)设置最大等待时长。import java.util.concurrent.Exchanger; public class Main { public static void main(String[] args) { final ExchangerString exgr new Exchanger(); new Thread(new Runnable() { Override public void run() { try { String A 你好我是AAA; String B exgr.exchange(A); System.out.println(我是线程 Thread.currentThread().getName() 我交换到了 B); } catch (Exception e) { throw new RuntimeException(e); } } }, A).start(); new Thread(new Runnable() { Override public void run() { try { String A 你好我是BBB; String B exgr.exchange(A); System.out.println(我是线程 Thread.currentThread().getName() 我交换到了 B); } catch (Exception e) { throw new RuntimeException(e); } } }, B).start();}}