**这是java高并发系列第15篇文章 **html
Semaphore(信号量)为多线程协做提供了更为强大的控制方法,前面的文章中咱们学了synchronized和重入锁ReentrantLock,这2种锁一次都只能容许一个线程访问一个资源,而信号量能够控制有多少个线程能够同时访问特定的资源。java
Semaphore经常使用场景:限流安全
举个例子:微信
好比有个停车场,有5个空位,门口有个门卫,手中5把钥匙分别对应5个车位上面的锁,来一辆车,门卫会给司机一把钥匙,而后进去找到对应的车位停下来,出去的时候司机将钥匙归还给门卫。停车场生意比较好,同时来了100两车,门卫手中只有5把钥匙,同时只能放5辆车进入,其余车只能等待,等有人将钥匙归还给门卫以后,才能让其余车辆进入。多线程
上面的例子中门卫就至关于Semaphore,车钥匙就至关于许可证,车就至关于线程。并发
Semaphore(int permits):构造方法,参数表示许可证数量,用来建立信号量异步
Semaphore(int permits,boolean fair):构造方法,当fair等于true时,建立具备给定许可数的计数信号量并设置为公平信号量分布式
void acquire() throws InterruptedException:今后信号量获取1个许可前线程将一直阻塞,至关于一辆车占了一个车位,此方法会响应线程中断,表示调用线程的interrupt方法,会使该方法抛出InterruptedException异常ide
void acquire(int permits) throws InterruptedException :和acquire()方法相似,参数表示须要获取许可的数量;好比一个大卡车要入停车场,因为车比较大,须要申请3个车位才能够停放高并发
void acquireUninterruptibly(int permits) :和acquire(int permits) 方法相似,只是不会响应线程中断
boolean tryAcquire():尝试获取1个许可,不论是否可以获取成功,都当即返回,true表示获取成功,false表示获取失败
boolean tryAcquire(int permits):和tryAcquire(),表示尝试获取permits个许可
boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException:尝试在指定的时间内获取1个许可,获取成功返回true,指定的时间事后仍是没法获取许可,返回false
boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException:和tryAcquire(long timeout, TimeUnit unit)相似,多了一个permits参数,表示尝试获取permits个许可
void release():释放一个许可,将其返回给信号量,至关于车从停车场出去时将钥匙归还给门卫
void release(int n):释放n个许可
int availablePermits():当前可用的许可数
package com.itsoku.chat12; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * 微信公众号:路人甲Java,专一于java技术分享(带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等),喜欢请关注! */ public class Demo1 { static Semaphore semaphore = new Semaphore(2); public static class T extends Thread { public T(String name) { super(name); } @Override public void run() { Thread thread = Thread.currentThread(); try { semaphore.acquire(); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可!"); TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); } finally { semaphore.release(); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",释放许可!"); } } } public static void main(String[] args) throws InterruptedException { for (int i = 0; i < 10; i++) { new T("t-" + i).start(); } } }
输出:
1563715791327,t-0,获取许可! 1563715791327,t-1,获取许可! 1563715794328,t-0,释放许可! 1563715794328,t-5,获取许可! 1563715794328,t-1,释放许可! 1563715794328,t-2,获取许可! 1563715797328,t-2,释放许可! 1563715797328,t-6,获取许可! 1563715797328,t-5,释放许可! 1563715797328,t-3,获取许可! 1563715800329,t-6,释放许可! 1563715800329,t-9,获取许可! 1563715800329,t-3,释放许可! 1563715800329,t-7,获取许可! 1563715803330,t-7,释放许可! 1563715803330,t-8,获取许可! 1563715803330,t-9,释放许可! 1563715803330,t-4,获取许可! 1563715806330,t-8,释放许可! 1563715806330,t-4,释放许可!
代码中new Semaphore(2)
建立了许可数量为2的信号量,每一个线程获取1个许可,同时容许两个线程获取许可,从输出中也能够看出,同时有两个线程能够获取许可,其余线程须要等待已获取许可的线程释放许可以后才能运行。为获取到许可的线程会阻塞在acquire()
方法上,直到获取到许可才能继续。
门卫(Semaphore)有点呆,司机进去的时候给了钥匙,出来的时候不归还,门卫也不会说什么。最终结果就是其余车辆都没法进入了。
以下代码:
package com.itsoku.chat12; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * 微信公众号:路人甲Java,专一于java技术分享(带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等),喜欢请关注! */ public class Demo2 { static Semaphore semaphore = new Semaphore(2); public static class T extends Thread { public T(String name) { super(name); } @Override public void run() { Thread thread = Thread.currentThread(); try { semaphore.acquire(); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可!"); TimeUnit.SECONDS.sleep(3); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",运行结束!"); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",当前可用许可数量:" + semaphore.availablePermits()); } catch (InterruptedException e) { e.printStackTrace(); } } } public static void main(String[] args) throws InterruptedException { for (int i = 0; i < 10; i++) { new T("t-" + i).start(); } } }
输出:
1563716603924,t-0,获取许可! 1563716603924,t-1,获取许可! 1563716606925,t-0,运行结束! 1563716606925,t-0,当前可用许可数量:0 1563716606925,t-1,运行结束! 1563716606925,t-1,当前可用许可数量:0
上面程序运行后一直没法结束,观察一下代码,代码中获取许可后,没有释放许可的代码,最终致使,可用许可数量为0,其余线程没法获取许可,会在semaphore.acquire();
处等待,致使程序没法结束。
示例1中,在finally里面释放锁,会有问题么?
若是获取锁的过程当中发生异常,致使获取锁失败,最后finally里面也释放了许可,最终会怎么样,致使许可数量凭空增加了。
示例代码:
package com.itsoku.chat12; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * 微信公众号:路人甲Java,专一于java技术分享(带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等),喜欢请关注! */ public class Demo3 { static Semaphore semaphore = new Semaphore(1); public static class T extends Thread { public T(String name) { super(name); } @Override public void run() { Thread thread = Thread.currentThread(); try { semaphore.acquire(); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可,当前可用许可数量:" + semaphore.availablePermits()); //休眠100秒 TimeUnit.SECONDS.sleep(100); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",运行结束!"); } catch (InterruptedException e) { e.printStackTrace(); } finally { semaphore.release(); } System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",当前可用许可数量:" + semaphore.availablePermits()); } } public static void main(String[] args) throws InterruptedException { T t1 = new T("t1"); t1.start(); //休眠1秒 TimeUnit.SECONDS.sleep(1); T t2 = new T("t2"); t2.start(); //休眠1秒 TimeUnit.SECONDS.sleep(1); T t3 = new T("t3"); t3.start(); //给t2和t3发送中断信号 t2.interrupt(); t3.interrupt(); } }
输出:
1563717279058,t1,获取许可,当前可用许可数量:0 java.lang.InterruptedException 1563717281060,t2,当前可用许可数量:1 at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) 1563717281060,t3,当前可用许可数量:2 at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.Semaphore.acquire(Semaphore.java:312) at com.itsoku.chat12.Demo3$T.run(Demo3.java:21) java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302) at java.util.concurrent.Semaphore.acquire(Semaphore.java:312) at com.itsoku.chat12.Demo3$T.run(Demo3.java:21)
程序中信号量许可数量为1,建立了3个线程获取许可,线程t1获取成功了,而后休眠100秒。其余两个线程阻塞在semaphore.acquire();
方法处,代码中对线程t二、t3发送中断信号,咱们看一下Semaphore中acquire的源码:
public void acquire() throws InterruptedException
这个方法会响应线程中断,主线程中对t二、t3发送中断信号以后,acquire()
方法会触发InterruptedException
异常,t二、t3最终没有获取到许可,可是他们都执行了finally中的释放许可的操做,最后致使许可数量变为了2,致使许可数量增长了。因此程序中释放许可的方式有问题。须要改进一下,获取许可成功才去释放锁。
正确的释放锁的方式,以下:
package com.itsoku.chat12; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * 微信公众号:路人甲Java,专一于java技术分享(带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等),喜欢请关注! */ public class Demo4 { static Semaphore semaphore = new Semaphore(1); public static class T extends Thread { public T(String name) { super(name); } @Override public void run() { Thread thread = Thread.currentThread(); //获取许但是否成功 boolean acquireSuccess = false; try { semaphore.acquire(); acquireSuccess = true; System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可,当前可用许可数量:" + semaphore.availablePermits()); //休眠100秒 TimeUnit.SECONDS.sleep(5); System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",运行结束!"); } catch (InterruptedException e) { e.printStackTrace(); } finally { if (acquireSuccess) { semaphore.release(); } } System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",当前可用许可数量:" + semaphore.availablePermits()); } } public static void main(String[] args) throws InterruptedException { T t1 = new T("t1"); t1.start(); //休眠1秒 TimeUnit.SECONDS.sleep(1); T t2 = new T("t2"); t2.start(); //休眠1秒 TimeUnit.SECONDS.sleep(1); T t3 = new T("t3"); t3.start(); //给t2和t3发送中断信号 t2.interrupt(); t3.interrupt(); } }
输出:
1563717751655,t1,获取许可,当前可用许可数量:0 1563717753657,t3,当前可用许可数量:0 java.lang.InterruptedException 1563717753657,t2,当前可用许可数量:0 at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302) at java.util.concurrent.Semaphore.acquire(Semaphore.java:312) at com.itsoku.chat12.Demo4$T.run(Demo4.java:23) java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) at java.util.concurrent.Semaphore.acquire(Semaphore.java:312) at com.itsoku.chat12.Demo4$T.run(Demo4.java:23) 1563717756656,t1,运行结束! 1563717756656,t1,当前可用许可数量:1
程序中增长了一个变量acquireSuccess
用来标记获取许但是否成功,在finally中根据这个变量是否为true,来肯定是否释放许可。
司机来到停车场,发现停车场已经满了,只能在外等待内部的车出来以后才能进去,可是要等多久,他本身也不知道,他但愿等10分钟,若是仍是没法进去,就不到这里停车了。
Semaphore内部2个方法能够提供超时获取许可的功能:
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException
在指定的时间内去尝试获取许可,若是可以获取到,返回true,获取不到返回false。
示例代码:
package com.itsoku.chat12; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * 微信公众号:路人甲Java,专一于java技术分享(带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等),喜欢请关注! */ public class Demo5 { static Semaphore semaphore = new Semaphore(1); public static class T extends Thread { public T(String name) { super(name); } @Override public void run() { Thread thread = Thread.currentThread(); //获取许但是否成功 boolean acquireSuccess = false; try { //尝试在1秒内获取许可,获取成功返回true,不然返回false System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",尝试获取许可,当前可用许可数量:" + semaphore.availablePermits()); acquireSuccess = semaphore.tryAcquire(1, TimeUnit.SECONDS); //获取成功执行业务代码 if (acquireSuccess) { System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可成功,当前可用许可数量:" + semaphore.availablePermits()); //休眠5秒 TimeUnit.SECONDS.sleep(5); } else { System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可失败,当前可用许可数量:" + semaphore.availablePermits()); } } catch (InterruptedException e) { e.printStackTrace(); } finally { if (acquireSuccess) { semaphore.release(); } } } } public static void main(String[] args) throws InterruptedException { T t1 = new T("t1"); t1.start(); //休眠1秒 TimeUnit.SECONDS.sleep(1); T t2 = new T("t2"); t2.start(); //休眠1秒 TimeUnit.SECONDS.sleep(1); T t3 = new T("t3"); t3.start(); } }
输出:
1563718410202,t1,尝试获取许可,当前可用许可数量:1 1563718410202,t1,获取许可成功,当前可用许可数量:0 1563718411203,t2,尝试获取许可,当前可用许可数量:0 1563718412203,t3,尝试获取许可,当前可用许可数量:0 1563718412204,t2,获取许可失败,当前可用许可数量:0 1563718413204,t3,获取许可失败,当前可用许可数量:0
代码中许可数量为1,semaphore.tryAcquire(1, TimeUnit.SECONDS);
:表示尝试在1秒内获取许可,获取成功当即返回true,超过1秒仍是获取不到,返回false。线程t1获取许可成功,以后休眠了5秒,从输出中能够看出t2和t3都尝试了1秒,获取失败。
throws InterruptedException
声明的,表示这个方法会响应线程中断信号,什么意思?表示调用线程的interrupt()
方法,会让这些方法触发InterruptedException
异常,即便这些方法处于阻塞状态,也会当即返回,并抛出InterruptedException
异常,线程中断信号也会被清除。java高并发系列连载中,总计估计会有四五十篇文章,能够关注公众号:javacode2018,送月薪3万课程,获取最新文章。
原文出处:https://www.cnblogs.com/itsoku123/p/11223837.html