Java 并发（6）Semaphore 源码分析

//构造器1
public Semaphore(int permits) {
   sync = new NonfairSync(permits);
}

//构造器2
public Semaphore(int permits, boolean fair) {
   sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

1、获取许可证

//获取一个许可证(响应中断)
public void acquire() throws InterruptedException {
   sync.acquireSharedInterruptibly(1);
}

//获取一个许可证(不响应中断)
public void acquireUninterruptibly() {
   sync.acquireShared(1);
}

//尝试获取许可证(非公平获取)
public boolean tryAcquire() {
   return sync.nonfairTryAcquireShared(1) >= 0;
}

//尝试获取许可证(定时获取)
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
   return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

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

abstract static class Sync extends AbstractQueuedSynchronizer {
   //非公平方式尝试获取
   final int nonfairTryAcquireShared(int acquires) {
       for (;;) {
           //获取可用许可证
           int available = getState();
           //获取剩余许可证
           int remaining = available - acquires;
           //1.如果remaining小于0则直接返回remaining
           //2.如果remaining大于0则先更新同步状态再返回remaining
           if (remaining < 0 || compareAndSetState(available, remaining)) {
               return remaining;
           }
       }
   }
}

//非公平同步器
static final class NonfairSync extends Sync {
   private static final long serialVersionUID = -2694183684443567898L;

   NonfairSync(int permits) {
       super(permits);
   }

   //尝试获取许可证
   protected int tryAcquireShared(int acquires) {
       return nonfairTryAcquireShared(acquires);
   }
}

//公平同步器
static final class FairSync extends Sync {
   private static final long serialVersionUID = 2014338818796000944L;

   FairSync(int permits) {
       super(permits);
   }

   //尝试获取许可证
   protected int tryAcquireShared(int acquires) {
       for (;;) {
           //判断同步队列前面有没有人排队
           if (hasQueuedPredecessors()) {
               //如果有的话就直接返回-1，表示尝试获取失败
               return -1;
           }
           //获取可用许可证
           int available = getState();
           //获取剩余许可证
           int remaining = available - acquires;
           //1.如果remaining小于0则直接返回remaining
           //2.如果remaining大于0则先更新同步状态再返回remaining
           if (remaining < 0 || compareAndSetState(available, remaining)) {
               return remaining;
           }
       }
   }
}

//以可中断模式获取锁(共享模式)
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
   //首先判断线程是否中断, 如果是则抛出异常
   if (Thread.interrupted()) {
       throw new InterruptedException();
   }
   //1.尝试去获取锁
   //负数：表示获取失败
   //零值：表示当前线程获取成功, 但是后继线程不能再获取了
   //正数：表示当前线程获取成功, 并且后继线程同样可以获取成功
   if (tryAcquireShared(arg) < 0) {
       //2. 如果获取失败则进人该方法
       doAcquireSharedInterruptibly(arg);
   }
}

2、释放许可证

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

// 释放锁的操作(共享模式)
public final boolean releaseShared(int arg) {
   //1.尝试去释放锁
   if (tryReleaseShared(arg)) {
       //2.如果释放成功就唤醒其他线程
       doReleaseShared();
       return true;
   }
   return false;
}

abstract static class Sync extends AbstractQueuedSynchronizer {
   ...
   //尝试释放操作
   protected final boolean tryReleaseShared(int releases) {
       for (;;) {
           //获取当前同步状态
           int current = getState();
           //将当前同步状态加上传入的参数
           int next = current + releases;
           //如果相加结果小于当前同步状态的话就报错
           if (next < current) {
               throw new Error("Maximum permit count exceeded");
           }
           //以CAS方式更新同步状态的值, 更新成功则返回true, 否则继续循环
           if (compareAndSetState(current, next)) {
               return true;
           }
       }
   }
   ...
}

3、动手写个连接池

public class ConnectPool {
   
   //连接池大小
   private int size;
   //数据库连接集合
   private Connect[] connects;
   //连接状态标志
   private boolean[] connectFlag;
   //剩余可用连接数
   private volatile int available;
   //信号量
   private Semaphore semaphore;
   
   //构造器
   public ConnectPool(int size) {
       this.size = size;
       this.available = size;
       semaphore = new Semaphore(size, true);
       connects = new Connect[size];
       connectFlag = new boolean[size];
       initConnects();
   }
   
   //初始化连接
   private void initConnects() {
       //生成指定数量的数据库连接
       for(int i = 0; i < this.size; i++) {
           connects[i] = new Connect();
       }
   }
   
   //获取数据库连接
   private synchronized Connect getConnect(){
       for(int i = 0; i < connectFlag.length; i++) {
           //遍历集合找到未使用的连接
           if(!connectFlag[i]) {
               //将连接设置为使用中
               connectFlag[i] = true;
               //可用连接数减1
               available--;
               System.out.println("【"+Thread.currentThread().getName()+"】以获取连接      剩余连接数：" + available);
               //返回连接引用
               return connects[i];
           }
       }
       return null;
   }
   
   //获取一个连接
   public Connect openConnect() throws InterruptedException {
       //获取许可证
       semaphore.acquire();
       //获取数据库连接
       return getConnect();
   }
   
   //释放一个连接
   public synchronized void release(Connect connect) {
       for(int i = 0; i < this.size; i++) {
           if(connect == connects[i]){
               //将连接设置为未使用
               connectFlag[i] = false;
               //可用连接数加1
               available++;
               System.out.println("【"+Thread.currentThread().getName()+"】以释放连接      剩余连接数：" + available);
               //释放许可证
               semaphore.release();
           }
       }
   }
   
   //剩余可用连接数
   public int available() {
       return available;
   }
   
}

测试代码：

public class TestThread extends Thread {
   
   private static ConnectPool pool = new ConnectPool(3);
   
   @Override
   public void run() {
       try {
           Connect connect = pool.openConnect();
           Thread.sleep(100);  //休息一下
           pool.release(connect);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }
   
   public static void main(String[] args) {
       for(int i = 0; i < 10; i++) {
           new TestThread().start();
       }
   }

}