专栏首页彤哥读源码死磕 java同步系列之Semaphore源码解析

死磕 java同步系列之Semaphore源码解析


问题

(1)Semaphore是什么?

(2)Semaphore具有哪些特性?

(3)Semaphore通常使用在什么场景中?

(4)Semaphore的许可次数是否可以动态增减?

(5)Semaphore如何实现限流?

简介

Semaphore,信号量,它保存了一系列的许可(permits),每次调用acquire()都将消耗一个许可,每次调用release()都将归还一个许可。

特性

Semaphore通常用于限制同一时间对共享资源的访问次数上,也就是常说的限流。

下面我们一起来学习Java中Semaphore是如何实现的。

类结构

Semaphore中包含了一个实现了AQS的同步器Sync,以及它的两个子类FairSync和NonFairSync,这说明Semaphore也是区分公平模式和非公平模式的。

源码分析

基于之前对于ReentrantLock和ReentrantReadWriteLock的分析,这篇文章相对来说比较简单,之前讲过的一些方法将直接略过,有兴趣的可以拉到文章底部查看之前的文章。

内部类Sync

// java.util.concurrent.Semaphore.Syncabstract static class Sync extends AbstractQueuedSynchronizer {    private static final long serialVersionUID = 1192457210091910933L;    // 构造方法,传入许可次数,放入state中    Sync(int permits) {        setState(permits);    }    // 获取许可次数    final int getPermits() {        return getState();    }    // 非公平模式尝试获取许可    final int nonfairTryAcquireShared(int acquires) {        for (;;) {            // 看看还有几个许可            int available = getState();            // 减去这次需要获取的许可还剩下几个许可            int remaining = available - acquires;            // 如果剩余许可小于0了则直接返回            // 如果剩余许可不小于0,则尝试原子更新state的值,成功了返回剩余许可            if (remaining < 0 ||                compareAndSetState(available, remaining))                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");            // 如果原子更新state的值成功,就说明释放许可成功,则返回true            if (compareAndSetState(current, next))                return true;        }    }    // 减少许可    final void reducePermits(int reductions) {        for (;;) {            // 看看还有几个许可            int current = getState();            // 减去将要减少的许可            int next = current - reductions;            // 检测举出            if (next > current) // underflow                throw new Error("Permit count underflow");            // 原子更新state的值,成功了返回true            if (compareAndSetState(current, next))                return;        }    }    // 销毁许可    final int drainPermits() {        for (;;) {            // 看看还有几个许可            int current = getState();            // 如果为0,直接返回            // 如果不为0,把state原子更新为0            if (current == 0 || compareAndSetState(current, 0))                return current;        }    }}

通过Sync的几个实现方法,我们获取到以下几点信息:

(1)许可是在构造方法时传入的;

(2)许可存放在状态变量state中;

(3)尝试获取一个许可的时候,则state的值减1;

(4)当state的值为0的时候,则无法再获取许可;

(5)释放一个许可的时候,则state的值加1;

(6)许可的个数可以动态改变;

内部类NonfairSync

// java.util.concurrent.Semaphore.NonfairSyncstatic final class NonfairSync extends Sync {    private static final long serialVersionUID = -2694183684443567898L;    // 构造方法,调用父类的构造方法    NonfairSync(int permits) {        super(permits);    }    // 尝试获取许可,调用父类的nonfairTryAcquireShared()方法    protected int tryAcquireShared(int acquires) {        return nonfairTryAcquireShared(acquires);    }}

非公平模式下,直接调用父类的nonfairTryAcquireShared()尝试获取许可。

内部类FairSync

// java.util.concurrent.Semaphore.FairSyncstatic final class FairSync extends Sync {    private static final long serialVersionUID = 2014338818796000944L;    // 构造方法,调用父类的构造方法    FairSync(int permits) {        super(permits);    }    // 尝试获取许可    protected int tryAcquireShared(int acquires) {        for (;;) {            // 公平模式需要检测是否前面有排队的            // 如果有排队的直接返回失败            if (hasQueuedPredecessors())                return -1;            // 没有排队的再尝试更新state的值            int available = getState();            int remaining = available - acquires;            if (remaining < 0 ||                compareAndSetState(available, remaining))                return remaining;        }    }}

公平模式下,先检测前面是否有排队的,如果有排队的则获取许可失败,进入队列排队,否则尝试原子更新state的值。

构造方法

// 构造方法,创建时要传入许可次数,默认使用非公平模式public Semaphore(int permits) {    sync = new NonfairSync(permits);}// 构造方法,需要传入许可次数,及是否公平模式public Semaphore(int permits, boolean fair) {    sync = fair ? new FairSync(permits) : new NonfairSync(permits);}

创建Semaphore时需要传入许可次数。

Semaphore默认也是非公平模式,但是你可以调用第二个构造方法声明其为公平模式。

下面的方法在学习过前面的内容看来都比较简单,彤哥这里只列举Semaphore支持的一些功能了。

以下的方法都是针对非公平模式来描述。

acquire()方法

public void acquire() throws InterruptedException {    sync.acquireSharedInterruptibly(1);}

获取一个许可,默认使用的是可中断方式,如果尝试获取许可失败,会进入AQS的队列中排队。

acquireUninterruptibly()方法

public void acquireUninterruptibly() {    sync.acquireShared(1);}

获取一个许可,非中断方式,如果尝试获取许可失败,会进入AQS的队列中排队。

tryAcquire()方法

public boolean tryAcquire() {    return sync.nonfairTryAcquireShared(1) >= 0;}

尝试获取一个许可,使用Sync的非公平模式尝试获取许可方法,不论是否获取到许可都返回,只尝试一次,不会进入队列排队。

tryAcquire(long timeout, TimeUnit unit)方法

public boolean tryAcquire(long timeout, TimeUnit unit)    throws InterruptedException {    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));}

尝试获取一个许可,先尝试一次获取许可,如果失败则会等待timeout时间,这段时间内都没有获取到许可,则返回false,否则返回true;

release()方法

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

释放一个许可,释放一个许可时state的值会加1,并且会唤醒下一个等待获取许可的线程。

acquire(int permits)方法

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

一次获取多个许可,可中断方式。

acquireUninterruptibly(int permits)方法

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

一次获取多个许可,非中断方式。

tryAcquire(int permits)方法

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

一次尝试获取多个许可,只尝试一次。

tryAcquire(int permits, long timeout, TimeUnit unit)方法

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

尝试获取多个许可,并会等待timeout时间,这段时间没获取到许可则返回false,否则返回true。

release(int permits)方法

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

一次释放多个许可,state的值会相应增加permits的数量。

availablePermits()方法

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

获取可用的许可次数。

drainPermits()方法

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

销毁当前可用的许可次数,对于已经获取的许可没有影响,会把当前剩余的许可全部销毁。

reducePermits(int reduction)方法

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

减少许可的次数。

总结

(1)Semaphore,也叫信号量,通常用于控制同一时刻对共享资源的访问上,也就是限流场景;

(2)Semaphore的内部实现是基于AQS的共享锁来实现的;

(3)Semaphore初始化的时候需要指定许可的次数,许可的次数是存储在state中;

(4)获取一个许可时,则state值减1;

(5)释放一个许可时,则state值加1;

(6)可以动态减少n个许可;

(7)可以动态增加n个许可吗?

彩蛋

(1)如何动态增加n个许可?

答:调用release(int permits)即可。我们知道释放许可的时候state的值会相应增加,再回头看看释放许可的源码,发现与ReentrantLock的释放锁还是有点区别的,Semaphore释放许可的时候并不会检查当前线程有没有获取过许可,所以可以调用释放许可的方法动态增加一些许可。

(2)如何实现限流?

答:限流,即在流量突然增大的时候,上层要能够限制住突然的大流量对下游服务的冲击,在分布式系统中限流一般做在网关层,当然在个别功能中也可以自己简单地来限流,比如秒杀场景,假如只有10个商品需要秒杀,那么,服务本身可以限制同时只进来100个请求,其它请求全部作废,这样服务的压力也不会太大。

使用Semaphore就可以直接针对这个功能来限流,以下是代码实现:

public class SemaphoreTest {    public static final Semaphore SEMAPHORE = new Semaphore(100);    public static final AtomicInteger failCount = new AtomicInteger(0);    public static final AtomicInteger successCount = new AtomicInteger(0);
    public static void main(String[] args) {        for (int i = 0; i < 1000; i++) {            new Thread(()->seckill()).start();        }    }
    public static boolean seckill() {        if (!SEMAPHORE.tryAcquire()) {            System.out.println("no permits, count="+failCount.incrementAndGet());            return false;        }
        try {            // 处理业务逻辑            Thread.sleep(2000);            System.out.println("seckill success, count="+successCount.incrementAndGet());        } catch (InterruptedException e) {            // todo 处理异常            e.printStackTrace();        } finally {            SEMAPHORE.release();        }        return true;    }}

本文分享自微信公众号 - 彤哥读源码(gh_63d1b83b9e01),作者:丹卿

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-06-16

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 死磕 Java集合之ArrayList源码分析

    ArrayList是一种以数组实现的List,与数组相比,它具有动态扩展的能力,因此也可称之为动态数组。

    彤哥
  • 死磕 java魔法类之Unsafe解析

    本章是java并发包专题的第一章,但是第一篇写的却不是java并发包中类,而是java中的魔法类sun.misc.Unsafe。

    彤哥
  • 死磕 java同步系列之Phaser源码解析

    (3)Phaser相对于CyclicBarrier和CountDownLatch的优势?

    彤哥
  • AQS源码分析之Semaphore

    公平版本在获取许可时需要先通过hasQueuedPredecessors方法判断是否有比当前节点等待更久的节点。

    开发架构二三事
  • Flink内存管理

    自从2003-2006年,Google发表了三篇著名的大数据相关论文(Google FS,MapReduce,Big Table)后,内存问题一直困扰大数据工程...

    神秘的寇先森
  • Java 字节流操作

         在java中我们使用输入流来向一个字节序列对象中写入,使用输出流来向输出其内容。C语言中只使用一个File包处理一切文件操作,而在java中却有着...

    Single
  • 2973 枪毙

    2973 枪毙 时间限制: 1 s 空间限制: 64000 KB 题目等级 : 青铜 Bronze 题目描述 Description   炼哥的朋...

    attack
  • 1833 深坑 TLE 求解

    题目描述:  大家知道,给出正整数n,则1到n这n个数可以构成n!种排列,把这些排列按照从小到大的顺序(字典顺序)列出,如n=3时,列出1 2 3,1 3 2,...

    瑾诺学长
  • 《你必须知道的.NET》读书实践:一个基于OO的万能加载器的实现

    此篇已收录至《你必须知道的.Net》读书笔记目录贴,点击访问该目录可以获取更多内容。

    Edison Zhou
  • 高效而稳定的企业级.NET Office 组件Spire(.NET组件介绍之二)

      在项目开发中,尤其是企业的业务系统中,对文档的操作是非常多的,有时几乎给人一种错觉的是”这个系统似乎就是专门操作文档的“。毕竟现在的很多办公中大都是在PC端...

    彭泽0902

扫码关注云+社区

领取腾讯云代金券