并发编程之Semaphore原理与应用

也想出现在这里？点击联系我~

前言

控制并发流程的工具类，作用就是帮助我们程序员更容易的让线程之间合作，让线程之间相互配合来满足业务逻辑。比如让线程A等待线程B执行完毕后再执行等合作策略。

控制并发流程的工具类主要有：

简介

Semaphore 信号量，许可，用于控制在一段时间内，可并发访问执行的线程数量。它的作用是控制访问特定资源的线程数目，底层依赖AQS的状态State，是在生产当中比较常用的一个工具类。

关于 AQS，可以查看《并发编程之抽象队列同步器AQS应用ReentrantLock》

一个信号量有且仅有 3 种操作，且它们全部是原子的。

• 初始化、增加和减少。
• 增加可以为一个进程解除阻塞。
• 减少可以让一个进程进入阻塞。

Semaphore 管理一系列许可证。

• 每个 acquire() 方法阻塞，直到有一个许可证可以获得然后拿走一个许可证。
• 每个 release() 方法增加一个许可证，这可能会释放一个阻塞的 acquire() 方法。
• 不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

Semaphore 在计数器不为 0 的时候对线程就放行，一旦达到 0，那么所有请求资源的新线程都会被阻塞，包括增加请求到许可的线程，Semaphore 是不可重入的。

• 每一次请求一个许可都会导致计数器减少 1，同样每次释放一个许可都会导致计数器增加 1，一旦达到 0，新的许可请求线程将被挂起。

Semaphore 有两种模式，公平模式 和 非公平模式 ，默认是非公平模式。

• 公平模式就是调用 acquire 的顺序就是获取许可证的顺序，遵循 FIFO。
• 非公平模式是抢占式的，也就是有可能一个新的获取线程恰好在一个许可证释放时得到了这个许可证，而前面还有等待的线程。

应用场景

Semaphore可以用来做流量限制，特别是公共资源有限的应用场景，比如说数据库连接。

由于 release() 释放许可时，未对释放许可数做限制，所有可以通过该方法增加总的许可数量; reducePermits() 方法可以减少总的许可数量，通过这两个方法可以到达动态调整许可的

分析：假如有一个需求，需读取几个万个文件的数据，因为都是IO密集型，我们可以启动几十个线程并发的读取，但是如果读取到内存后，还需要存储到数据库，而数据库的连接数只有10个，这时候我们就必须要控制只有10个线程同时获取到数据库连接，否则会抛出异常提示无法连接数据库。针对这种情况，我们就可以使用Semaphore来做流量控制。

代码如下：

1. packagecom.niuh.tools;
3. importjava.util.concurrent.ExecutorService;
4. importjava.util.concurrent.Executors;
5. importjava.util.concurrent.Semaphore;
6. importjava.util.concurrent.TimeUnit;
8. /**
9. *<p>
10. *Semaphore示例
11. *</p>
12. */
13. publicclassSemaphoreRunner{
14. /**
15. *线程数量
16. */
17. privatestaticfinalintTHREAD_COUNT=30;
19. /**
20. *线程池
21. */
22. privatestaticExecutorServiceexecutor=Executors.newFixedThreadPool(THREAD_COUNT);
24. privatestaticSemaphoresemaphore=newSemaphore(10);
26. publicstaticvoidmain(String[]args){
27. for(inti=0;i<THREAD_COUNT;i++){
28. executor.execute(newRunnable(){
29. publicvoidrun(){
30. try{
31. //获取一个"许可证"
32. semaphore.acquire();
34. //模拟数据保存
35. TimeUnit.SECONDS.sleep(2);
36. System.out.println("savedate…");
38. //执行完后,归还"许可证"
39. semaphore.release();
40. }catch(InterruptedExceptione){
41. e.printStackTrace();
42. }
43. }
44. });
45. }
46. executor.shutdown();
47. }
48. }

源码分析

Semaphore 类图

• Semaphore 通过使用内部类 Syn 继承 AQS 实现。

其内部主要变量和方法如下：

框架流程图如下：

构造函数

• permits 表示许可线程的数量
• fair 表示公平性，如果这个设为 true 的话，下次执行的线程会是等待最久的线程

1. publicSemaphore(intpermits){
2. sync=newNonfairSync(permits);
3. }
4. /**
5. *@parampermits总许可数
6. *@paramfairfair=true公平锁fair=false非公平锁
7. */
8. publicSemaphore(intpermits,booleanfair){
9. sync=fair?newFairSync(permits):newNonfairSync(permits);
10. }

内部类同步器

1. abstractstaticclassSyncextendsAbstractQueuedSynchronizer{
2. privatestaticfinallongserialVersionUID=1192457210091910933L;
3. //赋值setState为总许可数
4. Sync(intpermits){
5. setState(permits);
6. }
7. //剩余许可数
8. finalintgetPermits(){
9. returngetState();
10. }
11. //自旋+CAS非公平获取
12. finalintnonfairTryAcquireShared(intacquires){
13. for(;;){
14. //剩余可用许可数
15. intavailable=getState();
16. //本次获取许可后，剩余许可
17. intremaining=available-acquires;
18. //如果获取后，剩余许可大于0，则CAS更新剩余许可，否则获取更新失败
19. if(remaining<0||
20. compareAndSetState(available,remaining))
21. returnremaining;
22. }
23. }
24. //自旋+CAS释放许可
25. //由于未对释放许可数做限制，所以可以通过release动态增加许可数量
26. protectedfinalbooleantryReleaseShared(intreleases){
27. for(;;){
28. //当前剩余许可
29. intcurrent=getState();
30. //许可可更新值
31. intnext=current+releases;
32. //如果许可更新值为负数，说明许可数量益处，抛出错误
33. if(next<current)//overflow
34. thrownewError("Maximumpermitcountexceeded");
35. //CAS更新许可数量
36. if(compareAndSetState(current,next))
37. returntrue;
38. }
39. }
40. //自旋+CAS减少许可数量
41. finalvoidreducePermits(intreductions){
42. for(;;){
43. //当前剩余许可
44. intcurrent=getState();
45. //更新值
46. intnext=current-reductions;
47. //如果更新值比当前剩余许可大，抛出益处
48. if(next>current)//underflow
49. thrownewError("Permitcountunderflow");
50. //CAS更新许可数
51. if(compareAndSetState(current,next))
52. return;
53. }
54. }
55. //丢弃所有许可
56. finalintdrainPermits(){
57. for(;;){
58. intcurrent=getState();
59. if(current==0||compareAndSetState(current,0))
60. returncurrent;
61. }
62. }
63. }

非公平模式

1. /**
2. *非公平模式
3. */
4. staticfinalclassNonfairSyncextendsSync{
5. privatestaticfinallongserialVersionUID=-2694183684443567898L;
7. NonfairSync(intpermits){
8. super(permits);
9. }
11. protectedinttryAcquireShared(intacquires){
12. returnnonfairTryAcquireShared(acquires);
13. }
14. }

公平模式

1. /**
2. *公平模式
3. */
4. staticfinalclassFairSyncextendsSync{
5. privatestaticfinallongserialVersionUID=2014338818796000944L;
7. FairSync(intpermits){
8. super(permits);
9. }
10. //公平模式获取许可
11. //公平模式不论许可是否充足，都会判断同步队列中是否线程在等待，如果有，获取失败，排队阻塞
12. protectedinttryAcquireShared(intacquires){
13. for(;;){
14. //如果有线程在排队，立即返回
15. if(hasQueuedPredecessors())
16. return-1;
17. //自旋+CAS获取许可
18. intavailable=getState();
19. intremaining=available-acquires;
20. if(remaining<0||
21. compareAndSetState(available,remaining))
22. returnremaining;
23. }
24. }
25. }

获取许可

Semaphore 提供了两种获取资源的方式。

• 响应中断 和 不响应中断。

响应中断获取资源

两个方法支持 Interrupt 中断机制，可使用 acquire() 方法每次获取一个信号量，也可以使用 acquire(int permits) 方法获取指定数量的信号量 。

从semaphore中获取一个许可，线程会一直被阻塞直到获取一个许可或是被中断，获取一个许可后立即返回，并把许可数减1，如果没有可用的许可，当前线程会处于休眠状态直到：

1. 某些其他线程调用release方法，并且当前线程是下一个要被分配许可的线程
2. 某些其他线程中断当前线程

如果当前线程被acquire方法使得中断状态设置为on或者在等待许可时被中断则抛出InterruptedException，并且清除当前线程的已中断状态。

acquire执行流程：

1. publicvoidacquire()throwsInterruptedException{
2. sync.acquireSharedInterruptibly(1);
3. }
5. publicvoidacquire(intpermits)throwsInterruptedException{
6. if(permits<0)thrownewIllegalArgumentException();
7. sync.acquireSharedInterruptibly(permits);
8. }
11. publicfinalvoidacquireSharedInterruptibly(intarg)
12. throwsInterruptedException{
13. if(Thread.interrupted())
14. thrownewInterruptedException();
15. //获取许可，剩余许可>=0,则获取许可成功<0获取许可失败，进入排队
16. if(tryAcquireShared(arg)<0)
17. doAcquireSharedInterruptibly(arg);
18. }
20. /**
21. *获取许可失败，当前线程进入同步队列，排队阻塞
22. *@paramargtheacquireargument
23. */
24. privatevoiddoAcquireSharedInterruptibly(intarg)
25. throwsInterruptedException{
26. //创建同步队列节点，并入队列
27. finalNodenode=addWaiter(Node.SHARED);
28. booleanfailed=true;
29. try{
30. for(;;){
31. //如果当前节点是第二个节点，尝试获取锁
32. finalNodep=node.predecessor();
33. if(p==head){
34. intr=tryAcquireShared(arg);
35. if(r>=0){
36. setHeadAndPropagate(node,r);
37. p.next=null;//helpGC
38. failed=false;
39. return;
40. }
41. }
42. //阻塞当前线程
43. if(shouldParkAfterFailedAcquire(p,node)&&
44. parkAndCheckInterrupt())
45. thrownewInterruptedException();
46. }
47. }finally{
48. if(failed)
49. cancelAcquire(node);
50. }
51. }

代码的执行步骤如下：

AQS 子类使用共享模式，需要实现 tryAcquireShared() 方法。

1. 在公平锁中还是与ReentrantLock中的操作一样，先判断同步队列中是不是还有其他的等待线程，否则直接返回失败。否则对 state 值进行减操作并返回剩下的信号量。
2. 非公平锁直接调用了父类中的 nonfairTryAcquireShared 和 ReentrantLock 一样。

1. //非公平锁的获取方式
2. protectedinttryAcquireShared(intacquires){
3. returnnonfairTryAcquireShared(acquires);
4. }
7. finalintnonfairTryAcquireShared(intacquires){
8. for(;;){
9. intavailable=getState();//获取去中的信号量数
10. intremaining=available-acquires;//剩余信号量数
11. //1.信号量数大于0,获取共享锁,并设置执行compareAndSetState(available,remaining),返回剩余信号量数
12. //2.信号量数小于等于0,直接返回负数
13. if(remaining<0||compareAndSetState(available,remaining))
14. returnremaining;
15. }
16. }
18. //公平锁获取
19. protectedinttryAcquireShared(intacquires){
20. for(;;){
21. if(hasQueuedPredecessors())
22. return-1;
23. intavailable=getState();
24. intremaining=available-acquires;
25. if(remaining<0||compareAndSetState(available,remaining))
26. returnremaining;
27. }
28. }

变量 state 采用 volatile 可见修饰。

1. /**
2. *Thesynchronizationstate.
3. */
4. privatevolatileintstate;
6. /**
7. *Returnsthecurrentvalueofsynchronizationstate.
8. *Thisoperationhasmemorysemanticsofa<tt>volatile</tt>read.
9. *@returncurrentstatevalue
10. */
11. protectedfinalintgetState(){
12. returnstate;
13. }

不响应中断获取资源

两个方法不响应 Interrupt 中断机制，其它功能与 acquire() 方法一致。

从semaphore中获取一个许可，线程会一直被阻塞直到获取一个许可或是被中断，获取一个许可后立即返回，并把许可数减1，如果没有可用的许可，当前线程会处于休眠状态直到：

1. 某些其他线程调用release方法，并且当前线程是下一个要被分配许可的线程;
2. 如果当前线程在等待许可时被中断，那么它会接着等待，但是与没有发生中断相比，为线程分配许可的时间可能改变。

1. publicvoidacquireUninterruptibly(){
2. sync.acquireShared(1);
3. }
5. publicvoidacquireUninterruptibly(intpermits){
6. if(permits<0)thrownewIllegalArgumentException();
7. sync.acquireShared(permits);
8. }

尝试获得信号量

尝试获得信号量有三个方法。

• 尝试获取信号量，如果获取成功则返回 true，否则马上返回 false，不会阻塞当前线程。
• 尝试获取信号量，如果在指定的时间内获得信号量，则返回 true，否则返回 false。
• 尝试获取指定数量的信号量，如果在指定的时间内获得信号量，则返回 true，否则返回 false。

1. publicbooleantryAcquire(){
2. returnsync.nonfairTryAcquireShared(1)>=0;
3. }
5. publicbooleantryAcquire(longtimeout,TimeUnitunit)
6. throwsInterruptedException{
7. returnsync.tryAcquireSharedNanos(1,unit.toNanos(timeout));
8. }
10. publicbooleantryAcquire(intpermits,longtimeout,TimeUnitunit)
11. throwsInterruptedException{
12. if(permits<0)thrownewIllegalArgumentException();
13. returnsync.tryAcquireSharedNanos(permits,unit.toNanos(timeout));
14. }

释放归还许可

release 方法，主要作用是释放资源，需要保证 release 的执行，否则线程退出但是资源没有释放。

• 一般代码写在 finally 中是最好的。
• 并且获取多少资源就要释放多少资源，否则还是资源没被正确释放，如果一开始执行了 acquire(10) 最后释放的时候不能只写一个 release() 而是 release(10) 才对。

1. //尝试释放锁
2. publicfinalbooleanrelease(intarg){
3. //如果释放锁成功唤醒同步队列中的后继节点
4. if(tryRelease(arg)){
5. Nodeh=head;
6. if(h!=null&&h.waitStatus!=0)
7. unparkSuccessor(h);
8. returntrue;
9. }
10. returnfalse;
11. }
12. //为了方便对比把两个代码放在一块可以看到release中的结构完全一样
13. //区别就在于doReleaseShared中有更多的判断操作
14. publicfinalbooleanreleaseShared(intarg){
15. if(tryReleaseShared(arg)){
16. doReleaseShared();//在里面执行的unparkSuccessor(h)
17. returntrue;
18. }
19. returnfalse;
20. }

子类实现共享模式的类需要实现 tryReleaseShared() 方法判断是否释放成功。

• 这个方法是一个 CAS 自旋，原因是因为 Semaphore 是一个共享锁，可能有多个线程同时释放资源，因此 CAS 操作可能失败。

1. //由于未对释放许可数做限制，所以可以通过release动态增加许可数量
2. protectedfinalbooleantryReleaseShared(intreleases){
3. for(;;){
4. //获取当前许可数量
5. intcurrent=getState();
6. //计算回收后的数量
7. intnext=current+releases;
8. if(next<current)//overflow
9. thrownewError("Maximumpermitcountexceeded");
10. //CAS改变许可数量成功，返回true
11. if(compareAndSetState(current,next))
12. returntrue;
13. }
14. }

一旦 CAS 改变许可数量成功，就调用 doReleaseShared() 方法释放阻塞的线程。

1. privatevoiddoReleaseShared(){
2. //自旋，唤醒等待的第一个线程（其它线程将由第一个线程向后传递唤醒）
3. for(;;){
4. Nodeh=head;
5. if(h!=null&&h!=tail){
6. intws=h.waitStatus;
7. if(ws==Node.SIGNAL){
8. if(!compareAndSetWaitStatus(h,Node.SIGNAL,0))
9. continue;//looptorecheckcases
10. //唤醒第一个等待线程
11. unparkSuccessor(h);
12. }
13. elseif(ws==0&&
14. !compareAndSetWaitStatus(h,0,Node.PROPAGATE))
15. continue;//looponfailedCAS
16. }
17. if(h==head)//loopifheadchanged
18. break;
19. }
20. }

其他方法

获取当前剩余的信号量数量

• 该方法返回 AQS 中 state 变量的值，当前剩余的信号量个数。

1. publicintavailablePermits(){
2. returnsync.getPermits();
3. }
5. //Sync
6. finalintgetPermits(){
7. returngetState();
8. }

耗尽许可数量

• 获取并返回立即可用的所有许可。
• Sync 类的drainPermits()方法，获取 1 个信号量后将可用的信号量个数置为 0。例如总共有 10 个信号量，已经使用了 5 个，再调用 drainPermits() 方法后，可以获得一个信号量，剩余 4 个信号量就消失了，总共可用的信号量就变成 6 个了。用 CAS 自旋将剩余资源清空。

1. publicintdrainPermits(){
2. returnsync.drainPermits();
3. }
5. //Sync
6. finalintdrainPermits(){
7. for(;;){
8. intcurrent=getState();
9. if(current==0||compareAndSetState(current,0))
10. returncurrent;
11. }
12. }

缩减许可数量

• 缩减必须是单向的，即只能减少不能增加。用 CAS 自旋在剩余共享资源上做缩减。

1. protectedvoidreducePermits(intreduction){
2. if(reduction<0)thrownewIllegalArgumentException();
3. sync.reducePermits(reduction);
4. }
6. //Sync
7. finalvoidreducePermits(intreductions){
8. for(;;){
9. intcurrent=getState();
10. intnext=current-reductions;
11. if(next>current)//underflow
12. thrownewError("Permitcountunderflow");
13. if(compareAndSetState(current,next))
14. return;
15. }
16. }

上述两个方法对共享资源数量的修改操作有两点需要注意

• 是不可逆的
• 是对剩余资源的操作而不是全部资源，当剩余资源数目不足或已经为 0 时，方法就返回。
• 正在被占用的资源不参与。

判断 AQS 同步队列中是否还有 Node

1. publicfinalbooleanhasQueuedThreads(){
2. returnsync.hasQueuedThreads();
3. }
5. //AbstractQueuedSynchronizer
6. publicfinalbooleanhasQueuedThreads(){
7. //头结点不等于尾节点就说明链表中还有元素
8. returnhead!=tail;
9. }

总结

Semaphore 的内部工作流程也是基于 AQS，不同于 CyclicBarrier 和 ReentrantLock，不会使用到 AQS 的条件队列，都是在同步队列中操作，只是当前线程会被 park。

Semaphore 是 JUC 包提供的一个典型的共享锁，它通过自定义两种不同的同步器(FairSync 和 NonfairSync)提供了公平和非公平两种工作模式，两种模式下分别提供了限时/不限时、响应中断/不响应中断的获取资源的方法(限时获取总是及时响应中断的)，而所有的释放资源的 release() 操作是统一的。

PS：以上代码提交在 Github ：

https://github.com/Niuh-Study/niuh-juc-final.git