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的网络模型分为两部分:主线程和工作线程。主线程主要用来接收客户端的连接信息;工作线程主要用来接管客户