Semaphore源码分析

源码解析

Semaphore有两种模式,公平模式和非公平模式。公平模式就是调用acquire的顺序就是获取许可证的顺序,遵循FIFO;而非公平模式是抢占式的,也就是有可能一个新的获取线程恰好在一个许可证释放时得到了这个许可证,而前面还有等待的线程。

构造方法

Semaphore有两个构造方法,如下:

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

从上面可以看到两个构造方法,都必须提供许可的数量,第二个构造方法可以指定是公平模式还是非公平模式,默认非公平模式。
Semaphore内部基于AQS的共享模式,所以实现都委托给了Sync类。
这里就看一下NonfairSync的构造方法:

 NonfairSync(int permits) {            super(permits);
        }

可以看到直接调用了父类的构造方法,Sync的构造方法如下:

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

可以看到调用了setState方法,也就是说AQS中的资源就是许可证的数量。

获取许可

先从获取一个许可看起,并且先看非公平模式下的实现。首先看acquire方法,acquire方法有几个重载,但主要是下面这个方法

public void acquire(int permits) throws InterruptedException {        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }

从上面可以看到,调用了Sync的acquireSharedInterruptibly方法,该方法在父类AQS中,如下:

public final void acquireSharedInterruptibly(int arg)            throws InterruptedException {        //如果线程被中断了,抛出异常
        if (Thread.interrupted())            throw new InterruptedException();        //获取许可失败,将线程加入到等待队列中
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

AQS子类如果要使用共享模式的话,需要实现tryAcquireShared方法,下面看NonfairSync的该方法实现:

 protected int tryAcquireShared(int acquires) {            return nonfairTryAcquireShared(acquires);
        }

该方法调用了父类中的nonfairTyAcquireShared方法,如下:

final int nonfairTryAcquireShared(int acquires) {            for (;;) {                //获取剩余许可数量
                int available = getState();                //计算给完这次许可数量后的个数
                int remaining = available - acquires;                //如果许可不够或者可以将许可数量重置的话,返回
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))                    return remaining;
            }
        }

从上面可以看到,只有在许可不够时返回值才会小于0,其余返回的都是剩余许可数量,这也就解释了,一旦许可不够,后面的线程将会阻塞。看完了非公平的获取,再看下公平的获取,代码如下:

 protected int tryAcquireShared(int acquires) {            for (;;) {                //如果前面有线程再等待,直接返回-1
                if (hasQueuedPredecessors())                    return -1;                //后面与非公平一样
                int available = getState();                int remaining = available - acquires;                if (remaining < 0 ||
                    compareAndSetState(available, remaining))                    return remaining;
            }
        }

从上面可以看到,FairSync与NonFairSync的区别就在于会首先判断当前队列中有没有线程在等待,如果有,就老老实实进入到等待队列;而不像NonfairSync一样首先试一把,说不定就恰好获得了一个许可,这样就可以插队了。
看完了获取许可后,再看一下释放许可。

释放许可

释放许可也有几个重载方法,但都会调用下面这个带参数的方法,

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

releaseShared方法在AQS中,如下:

public final boolean releaseShared(int arg) {        //如果改变许可数量成功
        if (tryReleaseShared(arg)) {
            doReleaseShared();            return true;
        }        return false;
    }

AQS子类实现共享模式的类需要实现tryReleaseShared类来判断是否释放成功,实现如下:

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");                //CAS改变许可数量成功,返回true
                if (compareAndSetState(current, next))                    return true;
            }
        }

从上面可以看到,一旦CAS改变许可数量成功,那么就会调用doReleaseShared()方法释放阻塞的线程。

减小许可数量

Semaphore还有减小许可数量的方法,该方法可以用于用于当资源用完不能再用时,这时就可以减小许可证。代码如下:

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

可以看到,委托给了Sync,Sync的reducePermits方法如下:

  final void reducePermits(int reductions) {            for (;;) {                //得到当前剩余许可数量
                int current = getState();                //得到减完之后的许可数量
                int next = current - reductions;                if (next > current) // underflow
                    throw new Error("Permit count underflow");                //如果CAS改变成功
                if (compareAndSetState(current, next))                    return;
            }
        }

从上面可以看到,就是CAS改变AQS中的state变量,因为该变量代表许可证的数量。

获取剩余许可数量

Semaphore还可以一次将剩余的许可数量全部取走,该方法是drain方法,如下:

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

Sync的实现如下:

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

可以看到,就是CAS将许可数量置为0。

总结

Semaphore是信号量,用于管理一组资源。其内部是基于AQS的共享模式,AQS的状态表示许可证的数量,在许可证数量不够时,线程将会被挂起;而一旦有一个线程释放一个资源,那么就有可能重新唤醒等待队列中的线程继续执行。

原文地址:http://blog.51cto.com/8132260/2140813

时间: 2024-10-11 10:59:29

Semaphore源码分析的相关文章

Java - &quot;JUC&quot; Semaphore源码分析

Java多线程系列--"JUC锁"11之 Semaphore信号量的原理和示例 Semaphore简介 Semaphore是一个计数信号量,它的本质是一个"共享锁". 信号量维护了一个信号量许可集.线程可以通过调用acquire()来获取信号量的许可:当信号量中有可用的许可时,线程能获取该许可:否则线程必须等待,直到有可用的许可为止. 线程可以通过release()来释放它所持有的信号量许可. Java并发提供了两种加锁模式:共享锁和独占锁.前面LZ介绍的Reent

并发编程(七)——AbstractQueuedSynchronizer 之 CountDownLatch、CyclicBarrier、Semaphore 源码分析

这篇,我们的关注点是 AQS 最后的部分,共享模式的使用.本文先用 CountDownLatch 将共享模式说清楚,然后顺着把其他 AQS 相关的类 CyclicBarrier.Semaphore 的源码一起过一下. CountDownLatch CountDownLatch 这个类是比较典型的 AQS 的共享模式的使用,这是一个高频使用的类.使用方法在前面一篇文章中有介绍 并发编程(二)—— CountDownLatch.CyclicBarrier和Semaphore 使用例子 我们看下 Do

Semaphore 源码分析

Semaphore Semaphore 是基于同步器实现的计数信号量. Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源. 公平的信号量可以保证不会出现线程饥饿,而非公平的信号量可以提供更高的吞吐量. 创建实例 private final Sync sync; /** * 信号量的同步器实现 */ abstract static class Sync extends AbstractQueuedSynchronizer { priva

【JUC】JDK1.8源码分析之Semaphore(六)

一.前言 分析了CountDownLatch源码后,下面接着分析Semaphore的源码.Semaphore称为计数信号量,它允许n个任务同时访问某个资源,可以将信号量看做是在向外分发使用资源的许可证,只有成功获取许可证,才能使用资源.下面开始分析Semaphore的源码. 二.Semaphore的数据结构 分析源码可以知道,Semaphore底层是基于AbstractQueuedSynchronizer来实现的,所以,Semaphore的数据结构也依托于AQS的数据结构,在前面对AQS的分析中

Java并发包中Semaphore的工作原理、源码分析及使用示例

1. 信号量Semaphore的介绍 我们以一个停车场运作为例来说明信号量的作用.假设停车场只有三个车位,一开始三个车位都是空的.这时如果同时来了三辆车,看门人允许其中它们进入进入,然后放下车拦.以后来的车必须在入口等待,直到停车场中有车辆离开.这时,如果有一辆车离开停车场,看门人得知后,打开车拦,放入一辆,如果又离开一辆,则又可以放入一辆,如此往复. 在这个停车场系统中,车位是公共资源,每辆车好比一个线程,看门人起的就是信号量的作用.信号量是一个非负整数,表示了当前公共资源的可用数目(在上面的

【JDK】JDK源码分析-Semaphore

概述 Semaphore 是并发包中的一个工具类,可理解为信号量.通常可以作为限流器使用,即限制访问某个资源的线程个数,比如用于限制连接池的连接数. 打个通俗的比方,可以把 Semaphore 理解为一辆公交车:车上的座位数(初始的“许可” permits 数量)是固定的,行驶期间如果有人上车(获取许可),座位数(许可数量)就会减少,当人满的时候不能再继续上车了(获取许可失败):而有人下车(释放许可)后就空出了一些座位,其他人就可以继续上车了. 下面具体分析其代码实现. 代码分析 Semapho

Java显式锁学习总结之五:ReentrantReadWriteLock源码分析

概述 我们在介绍AbstractQueuedSynchronizer的时候介绍过,AQS支持独占式同步状态获取/释放.共享式同步状态获取/释放两种模式,对应的典型应用分别是ReentrantLock和Semaphore,AQS还可以混合两种模式使用,读写锁ReentrantReadWriteLock就是如此. 设想以下情景:我们在系统中有一个多线程访问的缓存,多个线程都可以对缓存进行读或写操作,但是读操作远远多于写操作,要求写操作要线程安全,且写操作执行完成要求对当前的所有读操作马上可见. 分析

JDK源码分析之concurrent包(一) -- Executor架构

Java5新出的concurrent包中的API,是一些并发编程中实用的的工具类.在高并发场景下的使用非常广泛.笔者在这做了一个针对concurrent包中部分常用类的源码分析系列.本系列针对的读者是已经对并发包中的Executor框架和工具类有所了解并懂得如何使用的人群,如果对并发包还不了解的朋友,请先做些了解.网上对这方面的讲述有丰富的资源. 本篇博文是第一期,首先对Executor架构做一个概述.这里只简单介绍接口和类的继承.使用关系. 盗用一张类图来描述结构: 解析: Executor是

《深入理解SPARK:核心思想与源码分析》——SparkContext的初始化(中)

<深入理解Spark:核心思想与源码分析>一书前言的内容请看链接<深入理解SPARK:核心思想与源码分析>一书正式出版上市 <深入理解Spark:核心思想与源码分析>一书第一章的内容请看链接<第1章 环境准备> <深入理解Spark:核心思想与源码分析>一书第二章的内容请看链接<第2章 SPARK设计理念与基本架构> 由于本书的第3章内容较多,所以打算分别开辟三篇随笔分别展现. <深入理解Spark:核心思想与源码分析>一