Java并发系列[6]----Semaphore源码分析

时间 2019-12-12

原文原文链接

Semaphore(信号量)是JUC包中比较经常使用到的一个类，它是AQS共享模式的一个应用，能够容许多个线程同时对共享资源进行操做，而且能够有效的控制并发数，利用它能够很好的实现流量控制。Semaphore提供了一个许可证的概念，能够把这个许可证看做公共汽车车票，只有成功获取车票的人才可以上车，而且车票是有必定数量的，不可能毫无限制的发下去，这样就会致使公交车超载。因此当车票发完的时候(公交车以满载)，其余人就只能等下一趟车了。若是中途有人下车，那么他的位置将会空闲出来，所以若是这时其余人想要上车的话就又能够得到车票了。利用Semaphore能够实现各类池，咱们在本篇末尾将会动手写一个简易的数据库链接池。首先咱们来看一下Semaphore的构造器。数据库

1 //构造器1
2 public Semaphore(int permits) {
3     sync = new NonfairSync(permits);
4 }
5 
6 //构造器2
7 public Semaphore(int permits, boolean fair) {
8     sync = fair ? new FairSync(permits) : new NonfairSync(permits);
9 }

Semaphore提供了两个带参构造器，没有提供无参构造器。这两个构造器都必须传入一个初始的许可证数量，使用构造器1构造出来的信号量在获取许可证时会采用非公平方式获取，使用构造器2能够经过参数指定获取许可证的方式(公平or非公平)。Semaphore主要对外提供了两类API，获取许可证和释放许可证，默认的是获取和释放一个许可证，也能够传入参数来同时获取和释放多个许可证。在本篇中咱们只讲每次获取和释放一个许可证的状况。数组

1.获取许可证并发

 1 //获取一个许可证(响应中断)
 2 public void acquire() throws InterruptedException {
 3     sync.acquireSharedInterruptibly(1);
 4 }
 5 
 6 //获取一个许可证(不响应中断)
 7 public void acquireUninterruptibly() {
 8     sync.acquireShared(1);
 9 }
10 
11 //尝试获取许可证(非公平获取)
12 public boolean tryAcquire() {
13     return sync.nonfairTryAcquireShared(1) >= 0;
14 }
15 
16 //尝试获取许可证(定时获取)
17 public 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源码系列文章的时候曾经讲过，如今咱们再来回顾一下。ide

 1 //以可中断模式获取锁(共享模式)
 2 public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
 3     //首先判断线程是否中断, 若是是则抛出异常
 4     if (Thread.interrupted()) {
 5         throw new InterruptedException();
 6     }
 7     //1.尝试去获取锁
 8     if (tryAcquireShared(arg) < 0) {
 9         //2. 若是获取失败则进人该方法
10         doAcquireSharedInterruptibly(arg);
11     }
12 }

acquireSharedInterruptibly方法首先就是去调用tryAcquireShared方法去尝试获取，tryAcquireShared在AQS里面是抽象方法，FairSync和NonfairSync这两个派生类实现了该方法的逻辑。FairSync实现的是公平获取的逻辑，而NonfairSync实现的非公平获取的逻辑。测试

 1 abstract static class Sync extends AbstractQueuedSynchronizer {
 2     //非公平方式尝试获取
 3     final int nonfairTryAcquireShared(int acquires) {
 4         for (;;) {
 5             //获取可用许可证
 6             int available = getState();
 7             //获取剩余许可证
 8             int remaining = available - acquires;
 9             //1.若是remaining小于0则直接返回remaining
10             //2.若是remaining大于0则先更新同步状态再返回remaining
11             if (remaining < 0 || compareAndSetState(available, remaining)) {
12                 return remaining;
13             }
14         }
15     }
16 }
17 
18 //非公平同步器
19 static final class NonfairSync extends Sync {
20     private static final long serialVersionUID = -2694183684443567898L;
21 
22     NonfairSync(int permits) {
23         super(permits);
24     }
25 
26     //尝试获取许可证
27     protected int tryAcquireShared(int acquires) {
28         return nonfairTryAcquireShared(acquires);
29     }
30 }
31 
32 //公平同步器
33 static final class FairSync extends Sync {
34     private static final long serialVersionUID = 2014338818796000944L;
35 
36     FairSync(int permits) {
37         super(permits);
38     }
39 
40     //尝试获取许可证
41     protected int tryAcquireShared(int acquires) {
42         for (;;) {
43             //判断同步队列前面有没有人排队
44             if (hasQueuedPredecessors()) {
45                 //若是有的话就直接返回-1，表示尝试获取失败
46                 return -1;
47             }
48             //获取可用许可证
49             int available = getState();
50             //获取剩余许可证
51             int remaining = available - acquires;
52             //1.若是remaining小于0则直接返回remaining
53             //2.若是remaining大于0则先更新同步状态再返回remaining
54             if (remaining < 0 || compareAndSetState(available, remaining)) {
55                 return remaining;
56             }
57         }
58     }
59 }

这里须要注意的是NonfairSync的tryAcquireShared方法直接调用的是nonfairTryAcquireShared方法，这个方法是在父类Sync里面的。非公平获取锁的逻辑是先取出当前同步状态(同步状态表示许可证个数)，将当前同步状态减去参入的参数，若是结果不小于0的话证实还有可用的许可证，那么就直接使用CAS操做更新同步状态的值，最后无论结果是否小于0都会返回该结果值。这里咱们要了解tryAcquireShared方法返回值的含义，返回负数表示获取失败，零表示当前线程获取成功但后续线程不能再获取，正数表示当前线程获取成功而且后续线程也可以获取。咱们再来看acquireSharedInterruptibly方法的代码。ui

 1 //以可中断模式获取锁(共享模式)
 2 public 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 }

若是返回的remaining小于0的话就表明获取失败，所以tryAcquireShared(arg) < 0就为true，因此接下来就会调用doAcquireSharedInterruptibly方法，这个方法咱们在讲AQS的时候讲过，它会将当前线程包装成结点放入同步队列尾部，而且有可能挂起线程。这也是当remaining小于0时线程会排队阻塞的缘由。而若是返回的remaining>=0的话就表明当前线程获取成功，所以tryAcquireShared(arg) < 0就为flase，因此就不会再去调用doAcquireSharedInterruptibly方法阻塞当前线程了。以上是非公平获取的整个逻辑，而公平获取时仅仅是在此以前先去调用hasQueuedPredecessors方法判断同步队列是否有人在排队，若是有的话就直接return -1表示获取失败，不然才继续执行下面和非公平获取同样的步骤。this

2.释放许可证spa

1 //释放一个许可证
2 public void release() {
3     sync.releaseShared(1);
4 }

调用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里面。code

 1 abstract 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 }

能够看到tryReleaseShared方法里面采用for循环进行自旋，首先获取同步状态，将同步状态加上传入的参数，而后以CAS方式更新同步状态，更新成功就返回true并跳出方法，不然就继续循环直到成功为止，这就是Semaphore释放许可证的流程。

3.动手写个链接池

Semaphore代码并无很复杂，经常使用的操做就是获取和释放一个许可证，这些操做的实现逻辑也都比较简单，但这并不妨碍Semaphore的普遍应用。下面咱们就来利用Semaphore实现一个简单的数据库链接池，经过这个例子但愿读者们能更加深刻的掌握Semaphore的运用。

 1 public 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 }

测试代码：

 1 public 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 }

测试结果：

咱们使用一个数组来存放数据库链接的引用，在初始化链接池的时候会调用initConnects方法建立指定数量的数据库链接，并将它们的引用存放到数组中，此外还有一个相同大小的数组来记录链接是否可用。每当外部线程请求获取一个链接时，首先调用semaphore.acquire()方法获取一个许可证，而后将链接状态设置为使用中，最后返回该链接的引用。许可证的数量由构造时传入的参数决定，每调用一次semaphore.acquire()方法许可证数量减1，当数量减为0时说明已经没有链接可使用了，这时若是其余线程再来获取就会被阻塞。每当线程释放一个链接的时候会调用semaphore.release()将许可证释放，此时许可证的总量又会增长，表明可用的链接数增长了，那么以前被阻塞的线程将会醒来继续获取链接，这时再次获取就可以成功获取链接了。测试示例中初始化了一个3个链接的链接池，咱们从测试结果中能够看到，每当线程获取一个链接剩余的链接数将会减1，等到减为0时其余线程就不能再获取了，此时必须等待一个线程将链接释放以后才能继续获取。能够看到剩余链接数老是在0到3之间变更，说明咱们此次的测试是成功的。