当前位置: 首页 > 编程笔记 >

JUC之Semaphore源码分析

陆涵畅
2023-03-14
本文向大家介绍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();
}

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小牛知识库。

 类似资料:
  • 主要内容:1.Semaphore 介绍,2.实例代码,3.资源获取acquire,4.释放资源1.Semaphore 介绍 Semaphore即信号量,常用于同时限制访问某些资源的线程数量。 其内部抽象类Fair继承了AQS,Semaphore正是通过Sync实现数量的控制 1.1 Sync Semaphore是基于AQS原理实现的,但并不是说Semaphore继承了AbstractQueuedSynchronizer抽象类,而是其内部类进行了AbstractQueuedSynchroni

  • ethereum的虚拟机源码所有部分在core/vm下。 去除测试总共有24个源码文件。 整个vm调用的入口在go-ethereum/core/state_transaction.go中。 我们主要是为了分析虚拟机源码,所以关于以太坊是如何进行交易转账忽略过去。 从上面的截图我们可以看出, 当以太坊的交易中to地址为nil时, 意味着部署合约, 那么就会调用evm.Create方法。 否则调用了e

  • 主要内容:1.ConcurrentSkipListMap概述,2.ConcurrentSkipListMap使用建议,3.SkipList 跳表介绍,4.ConcurrentSkipListMap,5.ConcurrentSkipListMap的查找,6.ConcurrentSkipListMap的删除,7.ConcurrentSkipListMap的插入1.ConcurrentSkipListMap概述 ConcurrentSkipListMap 的key是有序的。 ConcurrentSki

  • 主要内容:1.ConcurrentHashMap与HashMap等的区别,2.ConcurrentHashMap,3.总结我们熟知的缓存技术(比如redis、memcached)的核心其实就是在内存中维护一张巨大的哈希表,还有大家熟知的HashMap、CurrentHashMap等的应用。 1.ConcurrentHashMap与HashMap等的区别 1.1 HashMap 我们知道HashMap是线程不安全的,在多线程环境下,使用Hashmap进行put操作会引起死循环,导致CPU利用率接近

  • 区块链技术是计算机技术与金融技术交融的成功创新,被认为是极具潜力的分布式账本平台的核心技术。如果你还不了解区块链,可以阅读 区块链技术指南。

  • Memcached源码分析共8篇文章,前7篇文章主要分析每个模块的c源代码。这一篇文章主要是将之前的流程串起来,总结和回顾。同时通过这篇文章可以全局去看Memcached的结构。 一、Memcache的网络模型 Memcached主要是基于Libevent 网络事件库进行开发的。 Memcached的网络模型分为两部分:主线程和工作线程。主线程主要用来接收客户端的连接信息;工作线程主要用来接管客户