JUC之Semaphore源码分析

Semaphore 主要用于限量控制并发执行代码的工具类, 其内部通过 一个 permit 来进行定义并发执行的数量。

public KSemaphore(int permits){
 sync = new NonfairSync(permits);
}


public KSemaphore(int permits, boolean fair){
 sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

abstract static class Sync extends KAbstractQueuedSynchronizer{
 private static final long serialVersionUID = 1192457210091910933L;

 
 Sync(int permits){
  setState(permits);
 }

 
 final int getPermits(){
  return getState();
 }

 
 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;
   }
  }
 }

 
 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;
   }
  }
 }

 
 final int drainPermits(){
  for(;;){
   int current = getState();
   if(current == 0 || compareAndSetState(current, 0)){
    return current;
   }
  }
 }

}

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 void acquire(int permits) throws InterruptedException{
 if(permits < 0){
  throw new IllegalArgumentException();
 }
 sync.acquireSharedInterruptibly(permits);
}



public void acquireUninterruptibly(int permits){
 if(permits < 0) throw new IllegalArgumentException();
 sync.acquireShared(permits);
}



public boolean tryAcquire(int permits){
 if(permits < 0) throw new IllegalArgumentException();
 return sync.nonfairTryAcquireShared(permits) >= 0;
}



public boolean tryAcquire(int permits, long timout, TimeUnit unit) throws InterruptedException{
 if(permits < 0) throw new IllegalArgumentException();
 return sync.tryAcquireSharedNanos(permits, unit.tonanos(timout));
}

public void release(){
 sync.releaseShared(1);
}


public void release(int permits){
 if(permits < 0) throw new IllegalArgumentException();
 sync.releaseShared(permits);
}

public int availablePermits(){
 return sync.getPermits();
}



public int drainPermits(){
 return sync.drainPermits();
}



protected void reducePermits(int reduction){
 if(reduction < 0) throw new IllegalArgumentException();
 sync.reducePermits(reduction);
}



public boolean isFair(){
 return sync instanceof FairSync;
}



public final boolean hasQueuedThreads(){
 return sync.hasQueuedThreads();
}


public final int getQueueLength(){
 return sync.getQueueLength();
}



protected Collection getQueueThreads(){
 return sync.getQueuedThreads();
}

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持考高分网。