JUC之Semaphore源码分析

/** * 使用非公平版本构件 Semaphore */public KSemaphore(int permits){ sync = new NonfairSync(permits);}/** * 指定版本构件 Semaphore */public KSemaphore(int permits, boolean fair){ sync = fair ? new FairSync(permits) : new NonfairSync(permits);}

/** AQS 的子类主要定义获取释放 lock */abstract static class Sync extends KAbstractQueuedSynchronizer{ private static final long serialVersionUID = 1192457210091910933L; /**  * 指定 permit 初始化 Semaphore  */ Sync(int permits){  setState(permits); } /**  * 返回剩余 permit  */ final int getPermits(){  return getState(); } /**  * 获取 permit  */ final int nonfairTryAcquireShared(int acquires){  for(;;){   int available = getState();   int remaining = available - acquires; // 判断获取 acquires 的剩余 permit 数目   if(remaining < 0 ||     compareAndSetState(available, remaining)){ // cas改变 state    return remaining;   }  } } /**  * 释放 lock  */ protected final boolean tryReleaseShared(int releases){  for(;;){   int current = getState();   int next = current + releases;   if(next < current){ // overflow    throw new Error(" Maximum permit count exceeded");   }   if(compareAndSetState(current, next)){ // cas改变 state    return true;   }  } } final void reducePermits(int reductions){ // 减少 permits  for(;;){   int current = getState();   int next = current - reductions;   if(next > current){ // underflow    throw new Error(" Permit count underflow ");   }   if(compareAndSetState(current, next)){    return;   }  } } /** 将 permit 置为 0 */ final int drainPermits(){  for(;;){   int current = getState();   if(current == 0 || compareAndSetState(current, 0)){    return current;   }  } }}

/** * 调用 acquireSharedInterruptibly 响应中断的方式获取 permit */public void acquire() throws InterruptedException{ sync.acquireSharedInterruptibly(1);}/** * 调用 acquireUninterruptibly 非响应中断的方式获取 permit */public void acquireUninterruptibly(){ sync.acquireShared(1);}/** * 尝试获取 permit */public boolean tryAcquire(){ return sync.nonfairTryAcquireShared(1) >= 0;}/** * 尝试的获取 permit, 支持超时与中断 */public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException{ return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));}/** * 支持中断的获取permit */public void acquire(int permits) throws InterruptedException{ if(permits < 0){  throw new IllegalArgumentException(); } sync.acquireSharedInterruptibly(permits);}/** * 不响应中断的获取 permit */public void acquireUninterruptibly(int permits){ if(permits < 0) throw new IllegalArgumentException(); sync.acquireShared(permits);}/** * 尝试获取 permit */public boolean tryAcquire(int permits){ if(permits < 0) throw new IllegalArgumentException(); return sync.nonfairTryAcquireShared(permits) >= 0;}/** * 尝试 支持超时机制, 支持中断 的获取 permit */public boolean tryAcquire(int permits, long timout, TimeUnit unit) throws InterruptedException{ if(permits < 0) throw new IllegalArgumentException(); return sync.tryAcquireSharedNanos(permits, unit.toNanos(timout));}

/** * 释放 permit */public void release(){ sync.releaseShared(1);}/** * 释放 permit */public void release(int permits){ if(permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits);}

/** * 返回可用的 permit */public int availablePermits(){ return sync.getPermits();}/** * 消耗光 permit */public int drainPermits(){ return sync.drainPermits();}/** * 减少 reduction 个permit */protected void reducePermits(int reduction){ if(reduction < 0) throw new IllegalArgumentException(); sync.reducePermits(reduction);}/** * 判断是否是公平版本 */public boolean isFair(){ return sync instanceof FairSync;}/** * 返回 AQS 中 Sync Queue 里面的等待线程 */public final boolean hasQueuedThreads(){ return sync.hasQueuedThreads();}/** * 返回 AQS 中 Sync Queue 里面的等待线程长度 */public final int getQueueLength(){ return sync.getQueueLength();}/** * 返回 AQS 中 Sync Queue 里面的等待线程 */protected Collection<Thread> getQueueThreads(){ return sync.getQueuedThreads();}