JUC之Semaphore源码的示例分析

发布时间:2021-09-09 09:56:12 作者:小新
来源:亿速云 阅读:109

这篇文章主要为大家展示了“JUC之Semaphore源码的示例分析”,内容简而易懂,条理清晰,希望能够帮助大家解决疑惑,下面让小编带领大家一起研究并学习一下“JUC之Semaphore源码的示例分析”这篇文章吧。

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

/**
 * 使用非公平版本构件 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();
}

以上是“JUC之Semaphore源码的示例分析”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!

推荐阅读:
  1. JUC阻塞队列之DelayQueue源码分析
  2. 死磕 java同步系列之Semaphore源码解析

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

juc semaphore

上一篇:Linux 脚本编写基础知识以及在shell脚本中可以使用三类命令介绍

下一篇:怎么通过重启路由的方法切换IP地址

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》