一、Semaphore 信号量
1.1 API介绍
中间官方代码略。
Semaphore（信号量）是用来控制同时访问特定资源的线程数量，通过协调各个线程，保证合理的使用公共资源。
Semaphore维护了一个许可集，其实就是一定数量的“许可证”。
当有线程想要访问共享资源时，需要先获取(acquire)的许可；如果许可不够了，线程需要一直等待，直到许可可用。当线程使用完共享资源后，可以归还(release)许可，以供其它需要的线程使用。
和ReentrantLock类似，Semaphore支持公平/非公平策略。
1.2 源码简析
AQS共享模型
和ReentrantLock类很相似，底层也是sync，继承于AQS，不过ReentrantLock用的是独占模式，而Semaphore用的是共享模式：
public class Semaphore implements java.io.Serializable {
...
//获取
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
//释放
public void release() {
sync.releaseShared(1);
}
...
}
AQS共享模型需要实现的API：
- tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
- tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
...
//AQS共享模型尝试获取锁
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0) //获取许可,结果小于0则获取失败
doAcquireSharedInterruptibly(arg);//获取失败加入等待队列
}
//AQS共享模型释放锁
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
//尝试释放
doReleaseShared();//唤醒等待的线程
return true;
}
return false;
}
...
}
Java并发之AQS详解
AQS深入理解 doReleaseShared源码分析 JDK8
Semaphore中的实现
看下Semaphore的实现：
public class Semaphore implements java.io.Serializable {
...
/**
* NonFair version 非公平
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
NonfairSync(int permits) {
//初始信号量个数，父类会调用aqs的setState方法
super(permits);
}
protected int tryAcquireShared(int acquires) {
//nonfairTryAcquireShared方法实现在父类Sync中
return nonfairTryAcquireShared(acquires);
}
}
abstract static class Sync extends AbstractQueuedSynchronizer {
...
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();//获取许可个数
int remaining = available - acquires;//剩余的许可数 - 要获取的许可数
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;//不够返回负数，够返回剩余许可数
}
}
//释放
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
...
}
/**
* Fair version 公平
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L;
FairSync(int permits) {
super(permits);
}
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())//判断等待队列是否有线程等待
return -1;//有的话直接返回负数代表获取失败，当前线程会进入等待队列
//后面逻辑就和上面一样了
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
...
}
1.3 案例演示
/**
* 在信号量上我们定义两种操作：
* acquire（获取）当一个线程调用acquire操作时，他要么通过成功获取信号量（信号量减1），要么一直等待下去，直到有线程释放信号量，或超时。
* release（释放）实际上会将信号量加1，然后唤醒等待的线程。
*
* 信号量主要用于两个目的，一个是用于多个共享资源的互斥使用，另一个用于并发线程数的控制
*/
public class SemaphoreDemo {
public static void main(String[] args) {
Semaphore semaphore = new Semaphore(3);//模拟资源类，有3个空车位
for (int i = 1; i <= 6; i++) {
new Thread(()->{
try{
//占有资源
semaphore.acquire();
System.out.println(Thread.currentThread().getName()+"\t抢到车位");
try {
TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) {
e.printStackTrace(); }
System.out.println(Thread.currentThread().getName()+"\t停车3秒后离开车位");
} catch (Exception e) {
e.printStackTrace();
} finally {
//释放资源
semaphore.release();
}
}, String.valueOf(i)).start();
}
}
}
二、CountDownLatch 倒计时门闩
2.1 API介绍
CountDownLatch 内部维护了⼀个计数器，只有当计数器==0时，调用await的线程才会停⽌阻塞，开始执⾏。
官方使用案例1
下面是两个类，其中一组工作线程使用了两个倒计时锁存:
第一个是启动信号，阻止任何工人继续，直到司机准备好让他们继续;
class Driver {
// ...
void main() throws InterruptedException {
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(N);
for (int i = 0; i < N; ++i) // create and start threads
new Thread(new Worker(startSignal, doneSignal)).start();
doSomethingElse(); // don't let run yet
startSignal.countDown(); // let all threads proceed
doSomethingElse();
doneSignal.await(); // wait for all to finish
}
}
class Worker implements Runnable {
private final CountDownLatch startSignal;
private final CountDownLatch doneSignal;
Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
this.startSignal = startSignal;
this.doneSignal = doneSignal;
}
public void run() {
try {
startSignal.await();
doWork();
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() {
... }
}
官方使用案例2
第二个是一个完成信号，它允许驱动程序等待所有的工人完成。
另一个典型的用法是将一个问题分成N个部分，用一个运行程序来描述每个部分，执行该部分并在latch上倒数，然后将所有的运行程序排队到一个执行程序。当所有的子部分都完成时，协调线程将能够通过await。(当线程必须以这种方式重复计数时，请使用CyclicBarrier。)
class Driver2 {
// ...
void main() throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
Executor e = ...
for (int i = 0; i < N; ++i) // create and start threads
e.execute(new WorkerRunnable(doneSignal, i));
doneSignal.await(); // wait for all to finish
}
}
class WorkerRunnable implements Runnable {
private final CountDownLatch doneSignal;
private final int i;
WorkerRunnable(CountDownLatch doneSignal, int i) {
this.doneSignal = doneSignal;
this.i = i;
}
public void run() {
try {
doWork(i);
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() {
... }
}
在多线程协作完成业务功能时，有时候需要等待其他多个线程完成任务之后，主线程才能继续往下执行业务功能，在这种的业务场景下，通常可以使用Thread类的join方法，让主线程等待被join的线程执行完之后，主线程才能继续往下执行。当然，使用线程间消息通信机制也可以完成。其实，java并发工具类中为我们提供了类似“倒计时”这样的工具类，可以十分方便的完成所说的这种业务场景。
CountDownLatch允许一个或多个线程等待其他线程完成工作。
CountDownLatch相关方法：
- public CountDownLatch(int count) 构造方法会传入一个整型数N，之后调用CountDownLatch的 countDown 方法会对N减一，直到N减到0的时候，当前调用 await 方法的线程继续执行。
- await() throws InterruptedException：调用该方法的线程等到构造方法传入的N减到0的时候，或者被中断了，才能继续往下执行；
- await(long timeout, TimeUnit unit)：与上面的await方法功能一致，只不过这里有了时间限制，调用该方法的线程等到指定的timeout时间后，不管N是否减至为0，都会继续往下执行；
- countDown()：使CountDownLatch初始值N减1；
- long getCount()：获取当前CountDownLatch维护的值
2.2 源码简析
同理底层用的AQS共享模型
public class CountDownLatch {
...
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public void countDown() {
sync.releaseShared(1);
}
...
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
...
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);//获取锁失败则将当前线程加入等待队列
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
...
}
CountDownLatch中的实现
public class CountDownLatch {
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent count.
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
//只有计数为0才允许获取锁，否则调用await -> tryAcquireShared
//的线程会阻塞，进入阻塞队列
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// 每调用一次countDown，releases会传入1
// 然后计数-1
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;//只有0返回true，代表锁释放成功，才会唤醒等待的线程
}
}
}
...
}
2.3 案例演示
模拟如下场景：运动员进行跑步比赛时，假设有6个运动员参与比赛，裁判员在终点会为这6个运动员分别计时，可以想象没当一个运动员到达终点的时候，对于裁判员来说就少了一个计时任务。直到所有运动员都到达终点了，裁判员的任务也才完成。这6个运动员可以类比成6个线程，当线程调用CountDownLatch.countDown方法时就会对计数器的值减一，直到计数器的值为0的时候，裁判员（调用await方法的线程）才能继续往下执行。
public class CountDownLatchTest {
private static CountDownLatch startSignal = new CountDownLatch(1);
//用来表示裁判员需要维护的是6个运动员
private static CountDownLatch endSignal = new CountDownLatch(6);
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(6);
for (int i = 0; i < 6; i++) {
executorService.execute(() -> {
try {
System.out.println(Thread.currentThread().getName() + " 运动员等待裁判员响哨！！！");
startSignal.await();
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() + "正在全力冲刺");
endSignal.countDown();
System.out.println(Thread.currentThread().getName() + " 到达终点");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
System.out.println("裁判员响哨开始啦！！！");
startSignal.countDown();
endSignal.await();
System.out.println("所有运动员到达终点，比赛结束！");
executorService.shutdown();
}
}
该示例代码中设置了两个CountDownLatch，第一个endSignal用于控制让main线程（裁判员）必须等到其他线程（运动员）让CountDownLatch维护的数值N减到0为止，相当于一个完成信号；另一个startSignal用于让main线程对其他线程进行“发号施令”，相当于一个入口或者开关。
startSignal引用的CountDownLatch初始值为1，而其他线程执行的run方法中都会先通过startSignal.await()让这些线程都被阻塞，直到main线程通过调用startSignal.countDown();，将值N减1，CountDownLatch维护的数值N为0后，其他线程才能往下执行，并且，每个线程执行的run方法中都会通过endSignal.countDown();对endSignal维护的数值进行减一，由于往线程池提交了6个任务，会被减6次，所以endSignal维护的值最终会变为0，因此main线程在latch.await();阻塞结束，才能继续往下执行。
注意：当调用CountDownLatch的countDown方法时，当前线程是不会被阻塞，会继续往下执行。
