并发编程之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. package com.niuh.tools; 
2.  
3. import java.util.concurrent.ExecutorService; 
4. import java.util.concurrent.Executors; 
5. import java.util.concurrent.Semaphore; 
6. import java.util.concurrent.TimeUnit; 
7.  
8. /** 
9.  * <p> 
10.  * Semaphore示例 
11.  * </p> 
12.  */ 
13. public class SemaphoreRunner { 
14.     /** 
15.      * 线程数量 
16.      */ 
17.     private static final int THREAD_COUNT = 30; 
18.  
19.     /** 
20.      * 线程池 
21.      */ 
22.     private static ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT); 
23.  
24.     private static Semaphore semaphore = new Semaphore(10); 
25.  
26.     public static void main(String[] args) { 
27.         for (int i = 0; i < THREAD_COUNT; i++) { 
28.             executor.execute(new Runnable() { 
29.                 public void run() { 
30.                     try { 
31.                         // 获取一个"许可证" 
32.                         semaphore.acquire(); 
33.  
34.                         // 模拟数据保存 
35.                         TimeUnit.SECONDS.sleep(2); 
36.                         System.out.println("save date..."); 
37.  
38.                         // 执行完后,归还"许可证" 
39.                         semaphore.release(); 
40.                     } catch (InterruptedException e) { 
41.                         e.printStackTrace(); 
42.                     } 
43.                 } 
44.             }); 
45.         } 
46.         executor.shutdown(); 
47.     } 
48. } 

源码分析

Semaphore 类图

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

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

框架流程图如下：

构造函数

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

1. public Semaphore(int permits) { 
2.  sync = new NonfairSync(permits); 
3. } 
4. /** 
5. * @param permits 总许可数 
6. * @param fair fair=true 公平锁 fair=false 非公平锁 
7. */ 
8. public Semaphore(int permits, boolean fair) { 
9.  sync = fair ? new FairSync(permits) : new NonfairSync(permits); 
10. } 

内部类同步器

1. abstract static class Sync extends AbstractQueuedSynchronizer { 
2.     private static final long serialVersionUID = 1192457210091910933L; 
3.     // 赋值setState为总许可数 
4.     Sync(int permits) { 
5.         setState(permits); 
6.     } 
7.     // 剩余许可数 
8.     final int getPermits() { 
9.         return getState(); 
10.     } 
11.     // 自旋 + CAS 非公平获取 
12.     final int nonfairTryAcquireShared(int acquires) { 
13.         for (;;) { 
14.             // 剩余可用许可数 
15.             int available = getState(); 
16.             // 本次获取许可后，剩余许可 
17.             int remaining = available - acquires; 
18.             // 如果获取后，剩余许可大于0，则CAS更新剩余许可，否则获取更新失败 
19.             if (remaining < 0 || 
20.                 compareAndSetState(available, remaining)) 
21.                 return remaining; 
22.         } 
23.     } 
24.     // 自旋 + CAS 释放许可 
25.     // 由于未对释放许可数做限制，所以可以通过release动态增加许可数量 
26.     protected final boolean tryReleaseShared(int releases) { 
27.         for (;;) { 
28.             // 当前剩余许可 
29.             int current = getState(); 
30.             // 许可可更新值 
31.             int next = current + releases; 
32.             // 如果许可更新值为负数，说明许可数量益处，抛出错误 
33.             if (next < current) // overflow 
34.                 throw new Error("Maximum permit count exceeded"); 
35.             // CAS更新许可数量 
36.             if (compareAndSetState(current, next)) 
37.                 return true; 
38.         } 
39.     } 
40.     // 自旋 + CAS 减少许可数量 
41.     final void reducePermits(int reductions) { 
42.         for (;;) { 
43.             // 当前剩余许可 
44.             int current = getState(); 
45.             // 更新值 
46.             int next = current - reductions; 
47.             // 如果更新值比当前剩余许可大，抛出益处 
48.             if (next > current) // underflow 
49.                 throw new Error("Permit count underflow"); 
50.             // CAS 更新许可数 
51.             if (compareAndSetState(current, next)) 
52.                 return; 
53.         } 
54.     } 
55.     // 丢弃所有许可 
56.     final int drainPermits() { 
57.         for (;;) { 
58.             int current = getState(); 
59.             if (current == 0 || compareAndSetState(current, 0)) 
60.                 return current; 
61.         } 
62.     } 
63. } 

非公平模式

1. /** 
2. * 非公平模式 
3. */ 
4. static final class NonfairSync extends Sync { 
5.     private static final long serialVersionUID = -2694183684443567898L; 
6.  
7.     NonfairSync(int permits) { 
8.         super(permits); 
9.     } 
10.  
11.     protected int tryAcquireShared(int acquires) { 
12.         return nonfairTryAcquireShared(acquires); 
13.     } 
14. } 

公平模式

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

获取许可

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

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

响应中断获取资源

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

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

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

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

acquire执行流程：

1. public void acquire() throws InterruptedException { 
2.  sync.acquireSharedInterruptibly(1); 
3. } 
4.  
5. public void acquire(int permits) throws InterruptedException { 
6.     if (permits < 0) throw new IllegalArgumentException(); 
7.     sync.acquireSharedInterruptibly(permits); 
8. } 
9.  
10.  
11. public final void acquireSharedInterruptibly(int arg) 
12.         throws InterruptedException { 
13.     if (Thread.interrupted()) 
14.         throw new InterruptedException(); 
15.     // 获取许可，剩余许可 >= 0,则获取许可成功 <0 获取许可失败，进入排队 
16.     if (tryAcquireShared(arg) < 0) 
17.         doAcquireSharedInterruptibly(arg); 
18. } 
19.  
20. /** 
21.  * 获取许可失败，当前线程进入同步队列，排队阻塞 
22.  * @param arg the acquire argument 
23.  */ 
24. private void doAcquireSharedInterruptibly(int arg) 
25.     throws InterruptedException { 
26.     // 创建同步队列节点，并入队列 
27.     final Node node = addWaiter(Node.SHARED); 
28.     boolean failed = true; 
29.     try { 
30.         for (;;) { 
31.             // 如果当前节点是第二个节点，尝试获取锁 
32.             final Node p = node.predecessor(); 
33.             if (p == head) { 
34.                 int r = tryAcquireShared(arg); 
35.                 if (r >= 0) { 
36.                     setHeadAndPropagate(node, r); 
37.                     p.next = null; // help GC 
38.                     failed = false; 
39.                     return; 
40.                 } 
41.             } 
42.             // 阻塞当前线程 
43.             if (shouldParkAfterFailedAcquire(p, node) && 
44.                 parkAndCheckInterrupt()) 
45.                 throw new InterruptedException(); 
46.         } 
47.     } finally { 
48.         if (failed) 
49.             cancelAcquire(node); 
50.     } 
51. } 

代码的执行步骤如下：

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

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

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

变量 state 采用 volatile 可见修饰。

1. /** 
2.   * The synchronization state. 
3. */ 
4. private volatile int state; 
5.   
6. /** 
7.  * Returns the current value of synchronization state. 
8.  * This operation has memory semantics of a <tt>volatile</tt> read. 
9.  * @return current state value 
10. */ 
11. protected final int getState() { 
12.     return state; 
13. } 

不响应中断获取资源

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

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

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

1. public void acquireUninterruptibly() { 
2.     sync.acquireShared(1); 
3. } 
4.  
5. public void acquireUninterruptibly(int permits) { 
6.     if (permits < 0) throw new IllegalArgumentException(); 
7.     sync.acquireShared(permits); 
8. } 

尝试获得信号量

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

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

1. public boolean tryAcquire() { 
2.     return sync.nonfairTryAcquireShared(1) >= 0; 
3. } 
4.  
5. public boolean tryAcquire(long timeout, TimeUnit unit) 
6.         throws InterruptedException { 
7.     return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); 
8. } 
9.  
10. public boolean tryAcquire(int permits, long timeout, TimeUnit unit) 
11.         throws InterruptedException { 
12.     if (permits < 0) throw new IllegalArgumentException(); 
13.     return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout)); 
14. } 

释放归还许可

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

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

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

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

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

1. // 由于未对释放许可数做限制，所以可以通过release动态增加许可数量 
2. protected final boolean tryReleaseShared(int releases) { 
3.     for (;;) { 
4.         //获取当前许可数量 
5.         int current = getState(); 
6.         //计算回收后的数量 
7.         int next = current + releases; 
8.         if (next < current) // overflow 
9.             throw new Error("Maximum permit count exceeded"); 
10.         //CAS改变许可数量成功，返回true 
11.         if (compareAndSetState(current, next)) 
12.             return true; 
13.     } 
14. } 

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

1. private void doReleaseShared() { 
2.     // 自旋，唤醒等待的第一个线程（其它线程将由第一个线程向后传递唤醒） 
3.     for (;;) { 
4.         Node h = head; 
5.         if (h != null && h != tail) { 
6.             int ws = h.waitStatus; 
7.             if (ws == Node.SIGNAL) { 
8.                 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 
9.                     continue;            // loop to recheck cases 
10.                 // 唤醒第一个等待线程 
11.                 unparkSuccessor(h); 
12.             } 
13.             else if (ws == 0 && 
14.                      !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) 
15.                 continue;                // loop on failed CAS 
16.         } 
17.         if (h == head)                   // loop if head changed 
18.             break; 
19.     } 
20. } 

其他方法

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

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

1. public int availablePermits() { 
2.     return sync.getPermits(); 
3. } 
4.  
5. // Sync 
6. final int getPermits() { 
7.     return getState(); 
8. } 

耗尽许可数量

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

1. public int drainPermits() { 
2.     return sync.drainPermits(); 
3. } 
4.  
5. // Sync 
6. final int drainPermits() { 
7.     for (;;) { 
8.         int current = getState(); 
9.         if (current == 0 || compareAndSetState(current, 0)) 
10.             return current; 
11.     } 
12. } 

缩减许可数量

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

1. protected void reducePermits(int reduction) { 
2.     if (reduction < 0) throw new IllegalArgumentException(); 
3.     sync.reducePermits(reduction); 
4. } 
5.  
6. // Sync 
7. final void reducePermits(int reductions) { 
8.     for (;;) { 
9.         int current = getState(); 
10.         int next = current - reductions; 
11.         if (next > current) // underflow 
12.             throw new Error("Permit count underflow"); 
13.         if (compareAndSetState(current, next)) 
14.             return; 
15.     } 
16. } 

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

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

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

1. public final boolean hasQueuedThreads() { 
2.     return sync.hasQueuedThreads(); 
3. } 
4.  
5. // AbstractQueuedSynchronizer 
6. public final boolean hasQueuedThreads() { 
7.    //头结点不等于尾节点就说明链表中还有元素 
8.    return head != tail; 
9. } 

总结

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

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

PS：以上代码提交在 Github ：

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