详解Java 信号量Semaphore
Semaphore也是一个同步器，和前面两篇说的CountDownLatch和CyclicBarrier不同，这是递增的，初始化的时候可以指定一个值，但是不需要知道需要同步的线程个数，只需要在同步的地方调用acquire方法时指定需要同步的线程个数；
一.简单使用
同步两个子线程，只有其中两个子线程执行完毕，主线程才会执行：
packagecom.example.demo.study;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;
importjava.util.concurrent.Semaphore;
publicclassStudy0217{
//创建一个信号量的实例，信号量初始值为0
staticSemaphoresemaphore=newSemaphore(0);
publicstaticvoidmain(String[]args)throwsException{
ExecutorServicepool=Executors.newFixedThreadPool(3);
pool.submit(()->{
System.out.println("Thread1---start");
//信号量加一
semaphore.release();
});
pool.submit(()->{
System.out.println("Thread2---start");
//信号量加一
semaphore.release();
});
pool.submit(()->{
System.out.println("Thread3---start");
//信号量加一
semaphore.release();
});
//等待两个子线程执行完毕就放过，必须要信号量等于2才放过
semaphore.acquire(2);
System.out.println("两个子线程执行完毕");
//关闭线程池，正在执行的任务继续执行
pool.shutdown();
}
}
这个信号量也可以复用，类似CyclicBarrier：
packagecom.example.demo.study;
importjava.util.concurrent.ExecutorService;
importjava.util.concurrent.Executors;
importjava.util.concurrent.Semaphore;
publicclassStudy0217{
//创建一个信号量的实例，信号量初始值为0
staticSemaphoresemaphore=newSemaphore(0);
publicstaticvoidmain(String[]args)throwsException{
ExecutorServicepool=Executors.newFixedThreadPool(3);
pool.submit(()->{
System.out.println("Thread1---start");
//信号量加一
semaphore.release();
});
pool.submit(()->{
System.out.println("Thread2---start");
//信号量加一
semaphore.release();
});
//等待两个子线程执行完毕就放过，必须要信号量等于2才放过
semaphore.acquire(2);
System.out.println("子线程1，2执行完毕");
pool.submit(()->{
System.out.println("Thread3---start");
//信号量加一
semaphore.release();
});
pool.submit(()->{
System.out.println("Thread4---start");
//信号量加一
semaphore.release();
});
semaphore.acquire(2);
System.out.println("子线程3，4执行完毕");
//关闭线程池，正在执行的任务继续执行
pool.shutdown();
}
}
二.信号量原理
看看下面这个图，可以知道信号量Semaphore还是根据AQS实现的，内部有个Sync工具类操作AQS，还分为公平策略和非公平策略；
构造器：
//默认是非公平策略
publicSemaphore(intpermits){
sync=newNonfairSync(permits);
}
//可以根据第二个参数选择是公平策略还是非公平策略
publicSemaphore(intpermits,booleanfair){
sync=fair?newFairSync(permits):newNonfairSync(permits);
}
acquire(intpermits)方法：
publicvoidacquire(intpermits)throwsInterruptedException{
if(permits<0)thrownewIllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
//AQS中的方法
publicfinalvoidacquireSharedInterruptibly(intarg)
throwsInterruptedException{
if(Thread.interrupted())thrownewInterruptedException();
//这里根据子类是公平策略还是非公平策略
if(tryAcquireShared(arg)<0)
//获取失败会进入这里，将线程放入阻塞队列，然后再尝试，还是失败的话就调用park方法挂起当前线程
doAcquireSharedInterruptibly(arg);
}
//非公平策略
protectedinttryAcquireShared(intacquires){
returnnonfairTryAcquireShared(acquires);
}
finalintnonfairTryAcquireShared(intacquires){
//一个无限循环，获取state剩余的信号量，因为每调用一次release()方法的话，信号量就会加一，这里将
//最新的信号量减去传进来的参数比较，比如有两个线程，其中一个线程已经调用了release方法，然后调用acquire(2)方法，那么
//这里remaining的值就是-1,返回-1，然后当前线程就会被丢到阻塞队列中去了；如果另外一个线程也调用了release方法，
//那么此时的remaining==0，所以在这里的if中会调用CAS将0设置到state
//
for(;;){
intavailable=getState();
intremaining=available-acquires;
if(remaining<0||compareAndSetState(available,remaining))
returnremaining;
}
}
//公平策略
//和上面非公平差不多，只不过这里会查看阻塞队列中当前节点前面有没有前驱节点，有的话直接返回-1，
//就会把当前线程丢到阻塞队列中阻塞去了,没有前驱节点的话，就跟非公平模式一样的了
protectedinttryAcquireShared(intacquires){
for(;;){
if(hasQueuedPredecessors())
return-1;
intavailable=getState();
intremaining=available-acquires;
if(remaining<0||compareAndSetState(available,remaining))
returnremaining;
}
}
再看看release(intpermits)方法：
//这个方法的作用就是将信号量加一
publicvoidrelease(intpermits){
if(permits<0)thrownewIllegalArgumentException();
sync.releaseShared(permits);
}
//AQS中方法
publicfinalbooleanreleaseShared(intarg){
//tryReleaseShared尝试释放资源
if(tryReleaseShared(arg)){
//释放资源成功就调用park方法唤醒唤醒AQS队列中最前面的节点中的线程
doReleaseShared();
returntrue;
}
returnfalse;
}
protectedfinalbooleantryReleaseShared(intreleases){
//一个无限循环，获取state，然后加上传进去的参数，如果新的state的值小于旧的state，说明已经超过了state的最大值，溢出了
//没有溢出的话，就用CAS更新state的值
for(;;){
intcurrent=getState();
intnext=current+releases;
if(next
以最上面的例子简单说一下，其实不是很难，首先线程1和线程2分别去调用release方法，这个方法里面会将AQS中的state加一，但是在执行这个操作之前，主线程肯定会先到acquire(2)，在这个方法里面，假如默认使用非公平策略，首先获取当前的信号量state(state的初始值是0)，用当前信号量减去2,如果小于0，那么当前主线程就会丢到AQS队列中阻塞；
这个时候线程1的release方法执行了，于是就把信号量state加一(此时state==1)，CAS更新state为一，成功的话，就调用doReleaseShared()方法唤醒AQS阻塞队列中最先挂起的线程（这里就是因为调用acquire方法而阻塞的主线程），主线程唤醒之后又会去获取最新的信号量，与2比较，发现还是小于0，于是又会阻塞；
线程2此时的release方法执行完成，重复线程一的操作，主线程唤醒之后(此时state==2)，又去获取最新的信号量发现是2，减去acquire方法的参数2等于0，于是就用CAS更新state的值，然后acquire方法也就执行完毕，主线程继续执行后面的代码；
其实信号量还是很有意思的，记得在项目里，有人利用信号量实现了一个故障隔离，什么时候我可以把整理之后的代码贴出来分享一下，还是很有意思的，就跟springcloud的熔断机制差不多，场景是：比如你在service的一个方法调用第三方的接口，你不知道调不调得通，而且你不希望每次前端过来都会去调用，比如当调用失败的次数超过100次，那么五分钟之后才会再去实际调用这个第三方服务！这五分钟内前调用这个服务，就会触发我们这个故障隔离的机制，向前端返回一个特定的错误码和错误信息！
以上就是详解Java信号量Semaphore的详细内容，更多关于Java信号量Semaphore的资料请关注毛票票其它相关文章！
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。