Semaphore(信号量)是JUC包中比较常用到的一个类，它是AQS共享模式的一个应用，可以允许多个线程同时对共享资源进行操作，并且可以有效的控制并发数，利用它可以很好的实现流量控制。
Semaphore提供了一个许可证的概念，可以把这个许可证看作公共汽车车票，只有成功获取车票的人才能够上车，并且车票是有一定数量的，不可能毫无限制的发下去，这样就会导致公交车超载。所以当车票发完的时候(公交车以满载)，其他人就只能等下一趟车了。如果中途有人下车，那么他的位置将会空闲出来，因此如果这时其他人想要上车的话就又可以获得车票了。
利用Semaphore可以实现各种池，我们在本篇末尾将会动手写一个简易的数据库连接池。首先我们来看一下Semaphore的构造器。
1//构造器1 2public Semaphore(int permits) { 3 sync = new NonfairSync(permits); 4} 5 6//构造器2 7public Semaphore(int permits, boolean fair) { 8 sync = fair ? new FairSync(permits) : new NonfairSync(permits); 9} 10
Semaphore提供了两个带参构造器，没有提供无参构造器。这两个构造器都必须传入一个初始的许可证数量，使用构造器1构造出来的信号量在获取许可证时会采用非公平方式获取，使用构造器2可以通过参数指定获取许可证的方式(公平or非公平)。Semaphore主要对外提供了两类API，获取许可证和释放许可证，默认的是获取和释放一个许可证，也可以传入参数来同时获取和释放多个许可证。在本篇中我们只讲每次获取和释放一个许可证的情况。
1.获取许可证
1//获取一个许可证(响应中断) 2public void acquire() throws InterruptedException { 3 sync.acquireSharedInterruptibly(1); 4} 5 6//获取一个许可证(不响应中断) 7public void acquireUninterruptibly() { 8 sync.acquireShared(1); 9} 10 11//尝试获取许可证(非公平获取) 12public boolean tryAcquire() { 13 return sync.nonfairTryAcquireShared(1) >= 0; 14} 15 16//尝试获取许可证(定时获取) 17public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { 18 return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); 19}
上面的API是Semaphore提供的默认获取许可证操作。每次只获取一个许可证，这也是现实生活中较常遇到的情况。除了直接获取还提供了尝试获取，直接获取操作在失败之后可能会阻塞线程，而尝试获取则不会。
另外还需注意的是tryAcquire方法是使用非公平方式尝试获取的。在平时我们比较常用到的是acquire方法去获取许可证。下面我们就来看看它是怎样获取的。
可以看到acquire方法里面直接就是调用sync.acquireSharedInterruptibly(1)，这个方法是AQS里面的方法，我们在讲AQS源码系列文章的时候曾经讲过，现在我们再来回顾一下。
1//以可中断模式获取锁(共享模式) 2public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 3 //首先判断线程是否中断, 如果是则抛出异常 4 if (Thread.interrupted()) { 5 throw new InterruptedException(); 6 } 7 //1.尝试去获取锁 8 //负数：表示获取失败 9 //零值：表示当前线程获取成功, 但是后继线程不能再获取了 10 //正数：表示当前线程获取成功, 并且后继线程同样可以获取成功 11 if (tryAcquireShared(arg) < 0) { 12 //2. 如果获取失败则进人该方法 13 doAcquireSharedInterruptibly(arg); 14 } 15} 16 17protected int tryAcquireShared(int acquires) { 18 return nonfairTryAcquireShared(acquires); 19 } 20 21//非公平方式尝试获取 22 final int nonfairTryAcquireShared(int acquires) { 23 for (;;) { 24 //获取可用许可证 25 int available = getState(); 26 //获取剩余许可证 27 int remaining = available - acquires; 28 //1.如果remaining小于0则直接返回remaining 29 //2.如果remaining大于0则先更新同步状态再返回remaining 30 if (remaining < 0 || compareAndSetState(available, remaining)) 31 return remaining; 32 } 33 34//在同步队列中获取(共享模式) 35 private void doAcquireShared(int arg) { 36 //添加到同步队列中 37 final Node node = addWaiter(Node.SHARED); 38 boolean failed = true; 39 try { 40 boolean interrupted = false; 41 for (;;) { 42 //获取当前结点的前继结点 43 final Node p = node.predecessor(); 44 //如果前继结点为head结点就再次尝试去获取锁 45 if (p == head) { 46 //再次尝试去获取锁并返回获取状态 47 //r < 0, 表示获取失败 48 //r = 0, 表示当前结点获取成功, 但是后继结点不能再获取了 49 //r > 0, 表示当前结点获取成功, 并且后继结点同样可以获取成功 50 int r = tryAcquireShared(arg); 51 if (r >= 0) { 52 //到这里说明当前结点已经获取锁成功了, 此时它会将锁的状态信息传播给后继结点 53 setHeadAndPropagate(node, r); 54 p.next = null; // help GC 55 //如果在线程阻塞期间收到中断请求, 就在这一步响应该请求 56 if (interrupted) 57 selfInterrupt(); 58 failed = false; 59 return; 60 } 61 } 62 //每次获取锁失败后都会判断是否可以将线程挂起, 如果可以的话就会在parkAndCheckInterrupt方法里将线程挂起 63 if (shouldParkAfterFailedAcquire(p, node) && 64 parkAndCheckInterrupt()) 65 interrupted = true; 66 } 67 } finally { 68 if (failed) 69 cancelAcquire(node); 70 } 71 }
acquireSharedInterruptibly方法首先就是去调用tryAcquireShared方法去尝试获取，tryAcquireShared在AQS里面是抽象方法，FairSync和NonfairSync这两个派生类实现了该方法的逻辑。FairSync实现的是公平获取的逻辑，而NonfairSync实现的非公平获取的逻辑。
1//非公平同步器 2static final class NonfairSync extends Sync { 3 private static final long serialVersionUID = -2694183684443567898L; 4 5 NonfairSync(int permits) { 6 super(permits); 7 } 8 9 //尝试获取许可证 10 protected int tryAcquireShared(int acquires) { 11 return nonfairTryAcquireShared(acquires); 12 } 13} 14 15//公平同步器 16static final class FairSync extends Sync { 17 private static final long serialVersionUID = 2014338818796000944L; 18 19 FairSync(int permits) { 20 super(permits); 21 } 22 23 //尝试获取许可证 24 protected int tryAcquireShared(int acquires) { 25 for (;;) { 26 //判断同步队列前面有没有人排队 27 if (hasQueuedPredecessors()) { 28 //如果有的话就直接返回-1，表示尝试获取失败 29 return -1; 30 } 31 //获取可用许可证 32 int available = getState(); 33 //获取剩余许可证 34 int remaining = available - acquires; 35 //1.如果remaining小于0则直接返回remaining 36 //2.如果remaining大于0则先更新同步状态再返回remaining 37 if (remaining < 0 || compareAndSetState(available, remaining)) { 38 return remaining; 39 } 40 } 41 } 42} 43
这里需要注意的是NonfairSync的tryAcquireShared方法直接调用的是nonfairTryAcquireShared方法，这个方法是在父类Sync里面的。非公平获取锁的逻辑是先取出当前同步状态(同步状态表示许可证个数)，将当前同步状态减去参入的参数，如果结果不小于0的话证明还有可用的许可证，那么就直接使用CAS操作更新同步状态的值，最后不管结果是否小于0都会返回该结果值。
这里我们要了解tryAcquireShared方法返回值的含义，返回负数表示获取失败，零表示当前线程获取成功但后续线程不能再获取，正数表示当前线程获取成功并且后续线程也能够获取。
如果返回的remaining小于0的话就代表获取失败，因此tryAcquireShared(arg) < 0就为true，所以接下来就会调用doAcquireSharedInterruptibly方法，这个方法我们在讲AQS的时候讲过，它会将当前线程包装成结点放入同步队列尾部，并且有可能挂起线程。这也是当remaining小于0时线程会排队阻塞的原因。而如果返回的remaining>=0的话就代表当前线程获取成功，因此tryAcquireShared(arg) < 0就为flase，所以就不会再去调用doAcquireSharedInterruptibly方法阻塞当前线程了。
以上是非公平获取的整个逻辑，而公平获取时仅仅是在此之前先去调用hasQueuedPredecessors方法判断同步队列是否有人在排队，如果有的话就直接return -1表示获取失败，否则才继续执行下面和非公平获取一样的步骤。
2.释放许可证
1//释放一个许可证 2public void release() { 3 sync.releaseShared(1); 4} 5
调用release方法是释放一个许可证，它的操作很简单，就调用了AQS的releaseShared方法，我们来看看这个方法。
1//释放锁的操作(共享模式) 2 public final boolean releaseShared(int arg) { 3 //1.尝试去释放锁 4 if (tryReleaseShared(arg)) { 5 //2.如果释放成功就唤醒其他线程 6 doReleaseShared(); 7 return true; 8 } 9 return false; 10 }
AQS的releaseShared方法首先调用tryReleaseShared方法尝试释放锁，这个方法的实现逻辑在子类Sync里面。
1abstract static class Sync extends AbstractQueuedSynchronizer { 2 ... 3 //尝试释放操作 4 protected final boolean tryReleaseShared(int releases) { 5 for (;;) { 6 //获取当前同步状态 7 int current = getState(); 8 //将当前同步状态加上传入的参数 9 int next = current + releases; 10 //如果相加结果小于当前同步状态的话就报错 11 if (next < current) { 12 throw new Error("Maximum permit count exceeded"); 13 } 14 //以CAS方式更新同步状态的值, 更新成功则返回true, 否则继续循环 15 if (compareAndSetState(current, next)) { 16 return true; 17 } 18 } 19 } 20 ... 21} 22
可以看到tryReleaseShared方法里面采用for循环进行自旋，首先获取同步状态，将同步状态加上传入的参数，然后以CAS方式更新同步状态，更新成功就返回true并跳出方法，否则就继续循环直到成功为止，这就是Semaphore释放许可证的流程。
3.动手写个连接池
Semaphore代码并没有很复杂，常用的操作就是获取和释放一个许可证，这些操作的实现逻辑也都比较简单，但这并不妨碍Semaphore的广泛应用。下面我们就来利用Semaphore实现一个简单的数据库连接池，通过这个例子希望读者们能更加深入的掌握Semaphore的运用。
1public class ConnectPool { 2 3 //连接池大小 4 private int size; 5 //数据库连接集合 6 private Connect[] connects; 7 //连接状态标志 8 private boolean[] connectFlag; 9 //剩余可用连接数 10 private volatile int available; 11 //信号量 12 private Semaphore semaphore; 13 14 //构造器 15 public ConnectPool(int size) { 16 this.size = size; 17 this.available = size; 18 semaphore = new Semaphore(size, true); 19 connects = new Connect[size]; 20 connectFlag = new boolean[size]; 21 initConnects(); 22 } 23 24 //初始化连接 25 private void initConnects() { 26 //生成指定数量的数据库连接 27 for(int i = 0; i < this.size; i++) { 28 connects[i] = new Connect(); 29 } 30 } 31 32 //获取数据库连接 33 private synchronized Connect getConnect(){ 34 for(int i = 0; i < connectFlag.length; i++) { 35 //遍历集合找到未使用的连接 36 if(!connectFlag[i]) { 37 //将连接设置为使用中 38 connectFlag[i] = true; 39 //可用连接数减1 40 available--; 41 System.out.println("【"+Thread.currentThread().getName()+"】以获取连接 剩余连接数：" + available); 42 //返回连接引用 43 return connects[i]; 44 } 45 } 46 return null; 47 } 48 49 //获取一个连接 50 public Connect openConnect() throws InterruptedException { 51 //获取许可证 52 semaphore.acquire(); 53 //获取数据库连接 54 return getConnect(); 55 } 56 57 //释放一个连接 58 public synchronized void release(Connect connect) { 59 for(int i = 0; i < this.size; i++) { 60 if(connect == connects[i]){ 61 //将连接设置为未使用 62 connectFlag[i] = false; 63 //可用连接数加1 64 available++; 65 System.out.println("【"+Thread.currentThread().getName()+"】以释放连接 剩余连接数：" + available); 66 //释放许可证 67 semaphore.release(); 68 } 69 } 70 } 71 72 //剩余可用连接数 73 public int available() { 74 return available; 75 } 76 77} 78
测试代码：
1public class TestThread extends Thread { 2 3 private static ConnectPool pool = new ConnectPool(3); 4 5 @Override 6 public void run() { 7 try { 8 Connect connect = pool.openConnect(); 9 Thread.sleep(100); //休息一下 10 pool.release(connect); 11 } catch (InterruptedException e) { 12 e.printStackTrace(); 13 } 14 } 15 16 public static void main(String[] args) { 17 for(int i = 0; i < 10; i++) { 18 new TestThread().start(); 19 } 20 } 21 22} 23
测试结果：
我们使用一个数组来存放数据库连接的引用，在初始化连接池的时候会调用initConnects方法创建指定数量的数据库连接，并将它们的引用存放到数组中，此外还有一个相同大小的数组来记录连接是否可用。
每当外部线程请求获取一个连接时，首先调用semaphore.acquire()方法获取一个许可证，然后将连接状态设置为使用中，最后返回该连接的引用。许可证的数量由构造时传入的参数决定，每调用一次semaphore.acquire()方法许可证数量减1，当数量减为0时说明已经没有连接可以使用了，这时如果其他线程再来获取就会被阻塞。每当线程释放一个连接的时候会调用semaphore.release()将许可证释放，此时许可证的总量又会增加，代表可用的连接数增加了，那么之前被阻塞的线程将会醒来继续获取连接，这时再次获取就能够成功获取连接了。
测试示例中初始化了一个3个连接的连接池，我们从测试结果中可以看到，每当线程获取一个连接剩余的连接数将会减1，等到减为0时其他线程就不能再获取了，此时必须等待一个线程将连接释放之后才能继续获取。可以看到剩余连接数总是在0到3之间变动，说明我们这次的测试是成功的。
1/** 2 * 控制同时并发访问的线程个数，类似于有限的链表，常用语仅能提供有限个访问的资源,比如 控制数据库连接数 3 */ 4@Slf4j 5public class SemaphoreExample1 { 6 private static final Logger log = LoggerFactory.getLogger(SemaphoreExample1.class); 7 private final static int threadCount = 10; 8 public static void main(String[] args) throws Exception { 9 10 ExecutorService exec = Executors.newCachedThreadPool(); 11 12 final Semaphore semaphore = new Semaphore(3); 13 14 for (int i = 0; i < threadCount; i++) { 15 final int threadNum = i; 16 exec.execute(() -> { 17 try { 18 semaphore.acquire(); // 获取一个许可 19 test(threadNum); 20 semaphore.release(); // 释放一个许可 21 } catch (Exception e) { 22 log.error("exception", e); 23 } 24 }); 25 } 26 exec.shutdown(); 27 } 28 29 private static void test(int threadNum) throws Exception { 30 log.info("{}", threadNum); 31 Thread.sleep(1000); 32 } 33}
1public class SemaphoreExample2 { 2 private static final Logger log = LoggerFactory.getLogger(SemaphoreExample2.class); 3 private final static int threadCount = 10; 4 5 public static void main(String[] args) throws Exception { 6 7 ExecutorService exec = Executors.newCachedThreadPool(); 8 9 final Semaphore semaphore = new Semaphore(6); 10 11 for (int i = 0; i < threadCount; i++) { 12 final int threadNum = i; 13 exec.execute(() -> { 14 try { 15 semaphore.acquire(3); // 获取多个许可 16 test(threadNum); 17 semaphore.release(3); // 释放多个许可 18 } catch (Exception e) { 19 log.error("exception", e); 20 } 21 }); 22 } 23 exec.shutdown(); 24 } 25 26 private static void test(int threadNum) throws Exception { 27 log.info("{}", threadNum); 28 Thread.sleep(1000); 29 } 30} 31 32@Slf4j 33public class SemaphoreExample3 { 34 private static final Logger log = LoggerFactory.getLogger(SemaphoreExample3.class); 35 private final static int threadCount = 10; 36 37 public static void main(String[] args) throws Exception { 38 39 ExecutorService exec = Executors.newCachedThreadPool(); 40 41 final Semaphore semaphore = new Semaphore(3); 42 43 for (int i = 0; i < threadCount; i++) { 44 final int threadNum = i; 45 exec.execute(() -> { 46 try { 47 if (semaphore.tryAcquire()) { // 尝试获取一个许可 48 test(threadNum); 49 semaphore.release(); // 释放一个许可 50 }else { 51 test2(threadNum); 52 } 53 } catch (Exception e) { 54 log.error("exception", e); 55 } 56 }); 57 } 58 exec.shutdown(); 59 } 60 61 private static void test(int threadNum) throws Exception { 62 log.info("{}", threadNum); 63 Thread.sleep(1000); 64 } 65 private static void test2(int threadNum) throws Exception { 66 log.info("no-acquire,{}", threadNum); 67 Thread.sleep(1000); 68 } 69}
1@Slf4j 2public class SemaphoreExample4 { 3 private static final Logger log = LoggerFactory.getLogger(SemaphoreExample4.class); 4 private final static int threadCount = 20; 5 6 public static void main(String[] args) throws Exception { 7 8 ExecutorService exec = Executors.newCachedThreadPool(); 9 10 final Semaphore semaphore = new Semaphore(3); 11 12 for (int i = 0; i < threadCount; i++) { 13 final int threadNum = i; 14 exec.execute(() -> { 15 try { 16 if (semaphore.tryAcquire(5000, TimeUnit.MILLISECONDS)) { // 尝试获取一个许可 17 test(threadNum); 18 semaphore.release(); // 释放一个许可 19 }else{ 20 test2(threadNum); 21 } 22 } catch (Exception e) { 23 log.error("exception", e); 24 } 25 }); 26 } 27 exec.shutdown(); 28 } 29 30 private static void test(int threadNum) throws Exception { 31 log.info("{}", threadNum); 32 Thread.sleep(1000); 33 } 34 private static void test2(int threadNum) throws Exception { 35 log.info("no-acquire,{}", threadNum); 36 Thread.sleep(1000); 37 } 38}
