类 | 作用 |
---|---|
Semaphore | 限制线程的数量 |
Exchanger | 两个线程交换数据 |
CountDownLatch | 线程等待直到计数器减为 0 时开始工作 |
CyclicBarrier | 作用跟 CountDownLatch 类似,但是可以重复使用 |
Phaser | 增强的 CyclicBarrier |
Semaphore
Semaphore 翻译过来是信号的意思。顾名思义,这个工具类提供的功能就是多个线程彼此“传信号”。而这个“信号”是一个int
类型的数据,也可以看成是一种“资源”。
// 传入初始资源总数,默认情况下,是非公平的同步器
public Semaphore(int permits) {sync = new NonfairSync(permits);
}public Semaphore(int permits, boolean fair) {sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
最主要的方法是 acquire 方法和 release 方法。acquire()
方法会申请一个 permit,而 release 方法会释放一个 permit。当然,你也可以申请多个 acquire(int permits)
或者释放多个 release(int permits)
。
每次 acquire,permits 就会减少一个或者多个。如果减少到了 0,再有其他线程来 acquire,那就要阻塞这个线程直到有其它线程 release permit 为止。
Semaphore 往往用于资源有限的场景中,去限制线程的数量。举个例子,我想限制同时只能有 3 个线程在工作:
public class Main {static class MyThread implements Runnable {private final int value;private final Semaphore semaphore;public MyThread(int value, Semaphore semaphore) {this.value = value;this.semaphore = semaphore;}@Overridepublic void run() {try {// 获取 permitsemaphore.acquire();System.out.printf("当前线程是%d, 还剩%d个资源,还有%d个线程在等待%n",value, semaphore.availablePermits(), semaphore.getQueueLength());// 睡眠随机时间,打乱释放顺序Random random = new Random();Thread.sleep(random.nextInt(1000));System.out.printf("线程%d释放了资源%n", value);} catch (InterruptedException e) {e.printStackTrace();} finally {// 释放 permitsemaphore.release();}}}public static void main(String[] args) {Semaphore semaphore = new Semaphore(3);for (int i = 0; i < 10; i++) {new Thread(new MyThread(i, semaphore)).start();}}
}
Semaphore 默认的 acquire 方法是会让线程进入等待队列,且抛出异常中断。但它还有一些方法可以忽略中断或不进入阻塞队列:
// 忽略中断
public void acquireUninterruptibly()
public void acquireUninterruptibly(int permits)// 不进入等待队列,底层使用CAS
public boolean tryAcquire
public boolean tryAcquire(int permits)
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)throws InterruptedException
public boolean tryAcquire(long timeout, TimeUnit unit)
Semaphore 内部有一个继承了 AQS 的同步器 Sync,重写了tryAcquireShared
方法。在这个方法里,会去尝试获取资源。如果获取失败(想要的资源数量小于目前已有的资源数量),就会返回一个负数(代表尝试获取资源失败)。然后当前线程就会进入 AQS 的等待队列。
Exchanger
Exchanger 类用于两个线程交换数据。它支持泛型,也就是说可以在两个线程之间传送任何数据。
public class Main {public static void main(String[] args) throws InterruptedException {Exchanger<String> exchanger = new Exchanger<>();new Thread(() -> {try {System.out.println("这是线程A,得到了另一个线程的数据:"+ exchanger.exchange("这是来自线程A的数据"));} catch (InterruptedException e) {e.printStackTrace();}}).start();// 当一个线程调用 exchange 方法后,会处于阻塞状态,只有当另一个线程也调用了 exchange 方法,它才会继续执行System.out.println("这个时候线程A是阻塞的,在等待线程B的数据");Thread.sleep(1000);new Thread(() -> {try {System.out.println("这是线程B,得到了另一个线程的数据:"+ exchanger.exchange("这是来自线程B的数据"));} catch (InterruptedException e) {e.printStackTrace();}}).start();}
}
根据 JDK 里面注释的说法,可以总结为一下特性:
- 此类提供对外的操作是同步的;
- 用于成对出现的线程之间交换数据;
- 可以视作双向的同步队列;
- 可应用于基因算法、流水线设计等场景。
Exchanger 类还有一个有超时参数的方法,如果在指定时间内没有另一个线程调用 exchange,就会抛出一个超时异常。
public V exchange(V x, long timeout, TimeUnit unit)
Exchanger 是可以重复使用的。也就是说。两个线程可以使用 Exchanger 在内存中不断地再交换数据。
CountDownLatch
public class CountDownLatch {private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 4982264981922014374L;// 计数值(count)实际上就是闭锁需要等待的线程数量。这个值只能被设置一次,而且 CountDownLatch没有提供任何机制去重新设置这个计数值Sync(int count) {setState(count);}int getCount() {return getState();}protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}}}private final Sync sync;public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");this.sync = new Sync(count);}// 等待public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);}// 超时等待public boolean await(long timeout, TimeUnit unit)throws InterruptedException {return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));}// count - 1public void countDown() {sync.releaseShared(1);}// 获取当前还有多少countpublic long getCount() {return sync.getCount();}public String toString() {return super.toString() + "[Count = " + sync.getCount() + "]";}
}
public class Main {// 定义前置任务线程static class PreTaskThread implements Runnable {private final String task;private final CountDownLatch countDownLatch;public PreTaskThread(String task, CountDownLatch countDownLatch) {this.task = task;this.countDownLatch = countDownLatch;}@Overridepublic void run() {try {Random random = new Random();Thread.sleep(random.nextInt(1000));System.out.println(task + " - 任务完成");countDownLatch.countDown();} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) {// 假设有三个模块需要加载CountDownLatch countDownLatch = new CountDownLatch(3);// 主任务new Thread(() -> {try {System.out.println("等待数据加载...");System.out.printf("还有%d个前置任务%n", countDownLatch.getCount());countDownLatch.await();System.out.println("数据加载完成,正式开始游戏!");} catch (InterruptedException e) {e.printStackTrace();}}).start();// 前置任务new Thread(new PreTaskThread("加载地图数据", countDownLatch)).start();new Thread(new PreTaskThread("加载人物模型", countDownLatch)).start();new Thread(new PreTaskThread("加载背景音乐", countDownLatch)).start();}
}
CyclicBarrier
CyclicBarrirer 从名字上来理解是“循环屏障”的意思。前面提到了 CountDownLatch 一旦计数值count
被降为 0 后,就不能再重新设置了,它只能起一次“屏障”的作用。而 CyclicBarrier 拥有 CountDownLatch 的所有功能,还可以使用reset()
方法重置屏障。
如果参与者(线程)在等待的过程中,Barrier 被破坏,就会抛出 BrokenBarrierException。可以用isBroken()
方法检测 Barrier 是否被破坏。
- 如果有线程已经处于等待状态,调用 reset 方法会导致已经在等待的线程出现 BrokenBarrierException 异常。并且由于出现了 BrokenBarrierException,将会导致始终无法等待。
- 如果在等待的过程中,线程被中断,会抛出 InterruptedException 异常,并且这个异常会传播到其他所有的线程。
- 如果在执行屏障操作过程中发生异常,则该异常将传播到当前线程中,其他线程会抛出 BrokenBarrierException,屏障被损坏。
- 如果超出指定的等待时间,当前线程会抛出 TimeoutException 异常,其他线程会抛出 BrokenBarrierException 异常。
// 构造方法
public CyclicBarrier(int parties) {this(parties, null);
}public CyclicBarrier(int parties, Runnable barrierAction) {// 具体实现
}
public class Main {static class PreTaskThread implements Runnable {private final String task;private final CyclicBarrier cyclicBarrier;public PreTaskThread(String task, CyclicBarrier cyclicBarrier) {this.task = task;this.cyclicBarrier = cyclicBarrier;}@Overridepublic void run() {// 假设总共三个关卡for (int i = 1; i < 4; i++) {try {Random random = new Random();Thread.sleep(random.nextInt(1000));System.out.printf("关卡%d的任务%s完成%n", i, task);// 一旦调用 await 方法的线程数量等于构造方法中传入的任务总量(这里是 3),就代表达到屏障了。cyclicBarrier.await();} catch (InterruptedException | BrokenBarrierException e) {e.printStackTrace();}}}}public static void main(String[] args) {// CyclicBarrier 允许我们在达到屏障的时候可以执行一个任务,可以在构造方法传入一个 Runnable 类型的对象。CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {System.out.println("本关卡所有前置任务完成,开始游戏...");});new Thread(new PreTaskThread("加载地图数据", cyclicBarrier)).start();new Thread(new PreTaskThread("加载人物模型", cyclicBarrier)).start();new Thread(new PreTaskThread("加载背景音乐", cyclicBarrier)).start();}
}
CyclicBarrier 内部使用的是 Lock + Condition 实现的等待/通知模式。详情可以查看这个方法的源码:
private int dowait(boolean timed, long nanos)
Phaser
Phaser 是 Java 7 中引入的一个并发同步工具,它提供了对动态数量的线程的同步能力,这与 CyclicBarrier 和 CountDownLatch 不同,因为它们都需要预先知道等待的线程数量。Phaser 是多阶段的,意味着它可以同步不同阶段的多个操作。
前面我们介绍了 CyclicBarrier,可以发现它在构造方法里传入了“任务总量”parties
之后,就不能修改这个值了,并且每次调用await()
方法也只能消耗一个parties
计数。但 Phaser 可以动态地调整任务总量!
Phaser 是阶段性的,所以它有一个内部的阶段计数器。每当我们到达一个阶段的结尾时,Phaser 会自动前进到下一个阶段。
名词解释:
- Party:Phaser 的上下文中,一个 party 可以是一个线程,也可以是一个任务。当我们在 Phaser 上注册一个 party 时,Phaser 会递增它的参与者数量。
- arrive:对应一个 party 的状态,初始时是 unarrived,当调用
arriveAndAwaitAdvance()
或者arriveAndDeregister()
进入 arrive 状态,可以通过getUnarrivedParties()
获取当前未到达的数量。 - register:注册一个新的 party 到 Phaser。
- deRegister:减少一个 party。
- phase:阶段,当所有注册的 party 都 arrive 之后,将会调用 Phaser 的
onAdvance()
方法来判断是否要进入下一阶段。
Phaser 的终止有两种途径,Phaser 维护的线程执行完毕或者onAdvance()
返回true
。
public class Main {static class PreTaskThread implements Runnable {private final String task;private final Phaser phaser;public PreTaskThread(String task, Phaser phaser) {this.task = task;this.phaser = phaser;}@Overridepublic void run() {for (int i = 1; i < 4; i++) {try {// 从第二个关卡起,不加载新手教程if (i >= 2 && "加载新手教程".equals(task)) {continue;}Random random = new Random();Thread.sleep(random.nextInt(1000));System.out.printf("关卡%d,需要加载%d个模块,当前模块【%s】%n",i, phaser.getRegisteredParties(), task);// 从第二个关卡起,不加载新手教程if (i == 1 && "加载新手教程".equals(task)) {System.out.println("下次关卡移除加载【新手教程】模块");// 移除一个模块phaser.arriveAndDeregister();} else {phaser.arriveAndAwaitAdvance();}} catch (InterruptedException e) {e.printStackTrace();}}}}public static void main(String[] args) {Phaser phaser = new Phaser(4) {@Overrideprotected boolean onAdvance(int phase, int registeredParties) {System.out.printf("第%d次关卡准备完成%n", phase + 1);return phase == 3 || registeredParties == 0;}};new Thread(new PreTaskThread("加载地图数据", phaser)).start();new Thread(new PreTaskThread("加载人物模型", phaser)).start();new Thread(new PreTaskThread("加载背景音乐", phaser)).start();new Thread(new PreTaskThread("加载新手教程", phaser)).start();}
}
这里要注意关卡 1 的输出,在“加载新手教程”线程中调用了arriveAndDeregister()
减少一个 party 之后,后面的线程使用getRegisteredParties()
得到的是已经被修改后的 parties 了。但是当前这个阶段(phase),仍然是需要 4 个 parties 都 arrive 才触发屏障的。从下一个阶段开始,才需要 3 个 parties 都 arrive 就触发屏障。
Phaser 类用来控制某个阶段的线程数量很有用,但它并不在意这个阶段具体有哪些线程 arrive,只要达到它当前阶段的 parties 值,就触发屏障。所以我这里的案例虽然制定了特定的线程(加载新手教程)来更直观地表述 Phaser 的功能,但其实 Phaser 是没有分辨具体是哪个线程的功能的,它在意的只是数量。
它内部使用了两个基于 Fork-Join 框架的原子类辅助:
private final AtomicReference<QNode> evenQ;
private final AtomicReference<QNode> oddQ;static final class QNode implements ForkJoinPool.ManagedBlocker {// 实现代码
}