Java并发系列(6)Semaphore源码分析

Wesley13
• 阅读 280

Semaphore(信号量)是JUC包中比较常用到的一个类,它是AQS共享模式的一个应用,可以允许多个线程同时对共享资源进行操作,并且可以有效的控制并发数,利用它可以很好的实现流量控制。

Semaphore提供了一个许可证的概念,可以把这个许可证看作公共汽车车票,只有成功获取车票的人才能够上车,并且车票是有一定数量的,不可能毫无限制的发下去,这样就会导致公交车超载。所以当车票发完的时候(公交车以满载),其他人就只能等下一趟车了。如果中途有人下车,那么他的位置将会空闲出来,因此如果这时其他人想要上车的话就又可以获得车票了。

利用Semaphore可以实现各种池,我们在本篇末尾将会动手写一个简易的数据库连接池。首先我们来看一下Semaphore的构造器。

//构造器1
public Semaphore(int permits) {
   sync = new NonfairSync(permits);
}

//构造器2
public Semaphore(int permits, boolean fair) {
   sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
 

Semaphore提供了两个带参构造器,没有提供无参构造器。这两个构造器都必须传入一个初始的许可证数量,使用构造器1构造出来的信号量在获取许可证时会采用非公平方式获取,使用构造器2可以通过参数指定获取许可证的方式(公平or非公平)。Semaphore主要对外提供了两类API,获取许可证和释放许可证,默认的是获取和释放一个许可证,也可以传入参数来同时获取和释放多个许可证。在本篇中我们只讲每次获取和释放一个许可证的情况。

1.获取许可证

//获取一个许可证(响应中断)
public void acquire() throws InterruptedException {
   sync.acquireSharedInterruptibly(1);
}

//获取一个许可证(不响应中断)
public void acquireUninterruptibly() {
   sync.acquireShared(1);
}

//尝试获取许可证(非公平获取)
public boolean tryAcquire() {
   return sync.nonfairTryAcquireShared(1) >= 0;
}

//尝试获取许可证(定时获取)
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
   return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

上面的API是Semaphore提供的默认获取许可证操作。每次只获取一个许可证,这也是现实生活中较常遇到的情况。除了直接获取还提供了尝试获取,直接获取操作在失败之后可能会阻塞线程,而尝试获取则不会。

另外还需注意的是tryAcquire方法是使用非公平方式尝试获取的。在平时我们比较常用到的是acquire方法去获取许可证。下面我们就来看看它是怎样获取的。

可以看到acquire方法里面直接就是调用sync.acquireSharedInterruptibly(1),这个方法是AQS里面的方法,我们在讲AQS源码系列文章的时候曾经讲过,现在我们再来回顾一下。

//以可中断模式获取锁(共享模式)
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
   //首先判断线程是否中断, 如果是则抛出异常
   if (Thread.interrupted()) {
       throw new InterruptedException();
   }
   //1.尝试去获取锁
   //负数:表示获取失败
   //零值:表示当前线程获取成功, 但是后继线程不能再获取了
   //正数:表示当前线程获取成功, 并且后继线程同样可以获取成功
   if (tryAcquireShared(arg) < 0) {
       //2. 如果获取失败则进人该方法
       doAcquireSharedInterruptibly(arg);
   }
}

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

//非公平方式尝试获取
        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
            //获取可用许可证
            int available = getState();
            //获取剩余许可证
            int remaining = available - acquires;
            //1.如果remaining小于0则直接返回remaining
            //2.如果remaining大于0则先更新同步状态再返回remaining
            if (remaining < 0 || compareAndSetState(available, remaining))
                return remaining;
        }

//在同步队列中获取(共享模式)
    private void doAcquireShared(int arg) {
        //添加到同步队列中
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                //获取当前结点的前继结点
                final Node p = node.predecessor();
                //如果前继结点为head结点就再次尝试去获取锁
                if (p == head) {
                    //再次尝试去获取锁并返回获取状态
                    //r < 0, 表示获取失败
                    //r = 0, 表示当前结点获取成功, 但是后继结点不能再获取了
                    //r > 0, 表示当前结点获取成功, 并且后继结点同样可以获取成功
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        //到这里说明当前结点已经获取锁成功了, 此时它会将锁的状态信息传播给后继结点
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        //如果在线程阻塞期间收到中断请求, 就在这一步响应该请求
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                //每次获取锁失败后都会判断是否可以将线程挂起, 如果可以的话就会在parkAndCheckInterrupt方法里将线程挂起
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

acquireSharedInterruptibly方法首先就是去调用tryAcquireShared方法去尝试获取,tryAcquireShared在AQS里面是抽象方法,FairSync和NonfairSync这两个派生类实现了该方法的逻辑。FairSync实现的是公平获取的逻辑,而NonfairSync实现的非公平获取的逻辑。

//非公平同步器
static final class NonfairSync extends Sync {
   private static final long serialVersionUID = -2694183684443567898L;

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

   //尝试获取许可证
   protected int tryAcquireShared(int acquires) {
       return nonfairTryAcquireShared(acquires);
   }
}

//公平同步器
static final class FairSync extends Sync {
   private static final long serialVersionUID = 2014338818796000944L;

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

   //尝试获取许可证
   protected int tryAcquireShared(int acquires) {
       for (;;) {
           //判断同步队列前面有没有人排队
           if (hasQueuedPredecessors()) {
               //如果有的话就直接返回-1,表示尝试获取失败
               return -1;
           }
           //获取可用许可证
           int available = getState();
           //获取剩余许可证
           int remaining = available - acquires;
           //1.如果remaining小于0则直接返回remaining
           //2.如果remaining大于0则先更新同步状态再返回remaining
           if (remaining < 0 || compareAndSetState(available, remaining)) {
               return remaining;
           }
       }
   }
}
 

这里需要注意的是NonfairSync的tryAcquireShared方法直接调用的是nonfairTryAcquireShared方法,这个方法是在父类Sync里面的。非公平获取锁的逻辑是先取出当前同步状态(同步状态表示许可证个数),将当前同步状态减去参入的参数,如果结果不小于0的话证明还有可用的许可证,那么就直接使用CAS操作更新同步状态的值,最后不管结果是否小于0都会返回该结果值。

这里我们要了解tryAcquireShared方法返回值的含义,返回负数表示获取失败,零表示当前线程获取成功但后续线程不能再获取,正数表示当前线程获取成功并且后续线程也能够获取。

如果返回的remaining小于0的话就代表获取失败,因此tryAcquireShared(arg) < 0就为true,所以接下来就会调用doAcquireSharedInterruptibly方法,这个方法我们在讲AQS的时候讲过,它会将当前线程包装成结点放入同步队列尾部,并且有可能挂起线程。这也是当remaining小于0时线程会排队阻塞的原因。而如果返回的remaining>=0的话就代表当前线程获取成功,因此tryAcquireShared(arg) < 0就为flase,所以就不会再去调用doAcquireSharedInterruptibly方法阻塞当前线程了。

以上是非公平获取的整个逻辑,而公平获取时仅仅是在此之前先去调用hasQueuedPredecessors方法判断同步队列是否有人在排队,如果有的话就直接return -1表示获取失败,否则才继续执行下面和非公平获取一样的步骤。

2.释放许可证

//释放一个许可证
public void release() {
   sync.releaseShared(1);
}
 

调用release方法是释放一个许可证,它的操作很简单,就调用了AQS的releaseShared方法,我们来看看这个方法。

//释放锁的操作(共享模式)
    public final boolean releaseShared(int arg) {
        //1.尝试去释放锁
        if (tryReleaseShared(arg)) {
            //2.如果释放成功就唤醒其他线程
            doReleaseShared();
            return true;
        }
        return false;
    }

AQS的releaseShared方法首先调用tryReleaseShared方法尝试释放锁,这个方法的实现逻辑在子类Sync里面。

abstract static class Sync extends AbstractQueuedSynchronizer {
   ...
   //尝试释放操作
   protected final boolean tryReleaseShared(int releases) {
       for (;;) {
           //获取当前同步状态
           int current = getState();
           //将当前同步状态加上传入的参数
           int next = current + releases;
           //如果相加结果小于当前同步状态的话就报错
           if (next < current) {
               throw new Error("Maximum permit count exceeded");
           }
           //以CAS方式更新同步状态的值, 更新成功则返回true, 否则继续循环
           if (compareAndSetState(current, next)) {
               return true;
           }
       }
   }
   ...
}
 

可以看到tryReleaseShared方法里面采用for循环进行自旋,首先获取同步状态,将同步状态加上传入的参数,然后以CAS方式更新同步状态,更新成功就返回true并跳出方法,否则就继续循环直到成功为止,这就是Semaphore释放许可证的流程。

3.动手写个连接池

Semaphore代码并没有很复杂,常用的操作就是获取和释放一个许可证,这些操作的实现逻辑也都比较简单,但这并不妨碍Semaphore的广泛应用。下面我们就来利用Semaphore实现一个简单的数据库连接池,通过这个例子希望读者们能更加深入的掌握Semaphore的运用。

public class ConnectPool {
   
   //连接池大小
   private int size;
   //数据库连接集合
   private Connect[] connects;
   //连接状态标志
   private boolean[] connectFlag;
   //剩余可用连接数
   private volatile int available;
   //信号量
   private Semaphore semaphore;
   
   //构造器
   public ConnectPool(int size) {  
       this.size = size;
       this.available = size;
       semaphore = new Semaphore(size, true);
       connects = new Connect[size];
       connectFlag = new boolean[size];
       initConnects();
   }
   
   //初始化连接
   private void initConnects() {
       //生成指定数量的数据库连接
       for(int i = 0; i < this.size; i++) {
           connects[i] = new Connect();
       }
   }
   
   //获取数据库连接
   private synchronized Connect getConnect(){  
       for(int i = 0; i < connectFlag.length; i++) {
           //遍历集合找到未使用的连接
           if(!connectFlag[i]) {
               //将连接设置为使用中
               connectFlag[i] = true;
               //可用连接数减1
               available--;
               System.out.println("【"+Thread.currentThread().getName()+"】以获取连接      剩余连接数:" + available);
               //返回连接引用
               return connects[i];
           }
       }
       return null;
   }
   
   //获取一个连接
   public Connect openConnect() throws InterruptedException {
       //获取许可证
       semaphore.acquire();
       //获取数据库连接
       return getConnect();
   }
   
   //释放一个连接
   public synchronized void release(Connect connect) {  
       for(int i = 0; i < this.size; i++) {
           if(connect == connects[i]){
               //将连接设置为未使用
               connectFlag[i] = false;
               //可用连接数加1
               available++;
               System.out.println("【"+Thread.currentThread().getName()+"】以释放连接      剩余连接数:" + available);
               //释放许可证
               semaphore.release();
           }
       }
   }
   
   //剩余可用连接数
   public int available() {
       return available;
   }
   
}
 

测试代码:

public class TestThread extends Thread {
   
   private static ConnectPool pool = new ConnectPool(3);
   
   @Override
   public void run() {
       try {
           Connect connect = pool.openConnect();
           Thread.sleep(100);  //休息一下
           pool.release(connect);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
   }
   
   public static void main(String[] args) {
       for(int i = 0; i < 10; i++) {
           new TestThread().start();
       }
   }

}
 

测试结果:

Java并发系列(6)Semaphore源码分析

我们使用一个数组来存放数据库连接的引用,在初始化连接池的时候会调用initConnects方法创建指定数量的数据库连接,并将它们的引用存放到数组中,此外还有一个相同大小的数组来记录连接是否可用。

每当外部线程请求获取一个连接时,首先调用semaphore.acquire()方法获取一个许可证,然后将连接状态设置为使用中,最后返回该连接的引用。许可证的数量由构造时传入的参数决定,每调用一次semaphore.acquire()方法许可证数量减1,当数量减为0时说明已经没有连接可以使用了,这时如果其他线程再来获取就会被阻塞。每当线程释放一个连接的时候会调用semaphore.release()将许可证释放,此时许可证的总量又会增加,代表可用的连接数增加了,那么之前被阻塞的线程将会醒来继续获取连接,这时再次获取就能够成功获取连接了。

测试示例中初始化了一个3个连接的连接池,我们从测试结果中可以看到,每当线程获取一个连接剩余的连接数将会减1,等到减为0时其他线程就不能再获取了,此时必须等待一个线程将连接释放之后才能继续获取。可以看到剩余连接数总是在0到3之间变动,说明我们这次的测试是成功的。

/**
 * 控制同时并发访问的线程个数,类似于有限的链表,常用语仅能提供有限个访问的资源,比如 控制数据库连接数
 */
@Slf4j
public class SemaphoreExample1 {
    private static final Logger log = LoggerFactory.getLogger(SemaphoreExample1.class);
    private final static int threadCount = 10;
    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final Semaphore semaphore = new Semaphore(3);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire(); // 获取一个许可
                    test(threadNum);
                    semaphore.release(); // 释放一个许可
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

Java并发系列(6)Semaphore源码分析

public class SemaphoreExample2 {
    private static final Logger log = LoggerFactory.getLogger(SemaphoreExample2.class);
    private final static int threadCount = 10;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final Semaphore semaphore = new Semaphore(6);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire(3); // 获取多个许可
                    test(threadNum);
                    semaphore.release(3); // 释放多个许可
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

@Slf4j
public class SemaphoreExample3 {
    private static final Logger log = LoggerFactory.getLogger(SemaphoreExample3.class);
    private final static int threadCount = 10;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final Semaphore semaphore = new Semaphore(3);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    if (semaphore.tryAcquire()) { // 尝试获取一个许可
                        test(threadNum);
                        semaphore.release(); // 释放一个许可
                        }else {
                        test2(threadNum);
                    }
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
    private static void test2(int threadNum) throws Exception {
        log.info("no-acquire,{}", threadNum);
        Thread.sleep(1000);
    }
}

Java并发系列(6)Semaphore源码分析

@Slf4j
public class SemaphoreExample4 {
    private static final Logger log = LoggerFactory.getLogger(SemaphoreExample4.class);
    private final static int threadCount = 20;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final Semaphore semaphore = new Semaphore(3);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    if (semaphore.tryAcquire(5000, TimeUnit.MILLISECONDS)) { // 尝试获取一个许可
                        test(threadNum);
                        semaphore.release(); // 释放一个许可
                    }else{
                        test2(threadNum);
                    }
                } catch (Exception e) {
                    log.error("exception", e);
                }
            });
        }
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
    private static void test2(int threadNum) throws Exception {
        log.info("no-acquire,{}", threadNum);
        Thread.sleep(1000);
    }
}

Java并发系列(6)Semaphore源码分析

点赞
收藏
评论区
推荐文章
秃头王路飞 秃头王路飞
4个月前
webpack5手撸vue2脚手架
webpack5手撸vue相信工作个12年的小伙伴们在面试的时候多多少少怕被问到关于webpack方面的知识,本菜鸟最近闲来无事,就尝试了手撸了下vue2的脚手架,第一次发帖实在是没有经验,望海涵。languageJavaScript"name":"vuecliversion2","version":"1.0.0","desc
技术小男生 技术小男生
4个月前
linux环境jdk环境变量配置
1:编辑系统配置文件vi/etc/profile2:按字母键i进入编辑模式,在最底部添加内容:JAVAHOME/opt/jdk1.8.0152CLASSPATH.:$JAVAHOME/lib/dt.jar:$JAVAHOME/lib/tools.jarPATH$JAVAHOME/bin:$PATH3:生效配置
光头强的博客 光头强的博客
4个月前
Java面向对象试题
1、请创建一个Animal动物类,要求有方法eat()方法,方法输出一条语句“吃东西”。创建一个接口A,接口里有一个抽象方法fly()。创建一个Bird类继承Animal类并实现接口A里的方法输出一条有语句“鸟儿飞翔”,重写eat()方法输出一条语句“鸟儿吃虫”。在Test类中向上转型创建b对象,调用eat方法。然后向下转型调用eat()方
刚刚好 刚刚好
4个月前
css问题
1、在IOS中图片不显示(给图片加了圆角或者img没有父级)<div<imgsrc""/</divdiv{width:20px;height:20px;borderradius:20px;overflow:h
blmius blmius
1年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
小森森 小森森
4个月前
校园表白墙微信小程序V1.0 SayLove -基于微信云开发-一键快速搭建,开箱即用
后续会继续更新,敬请期待2.0全新版本欢迎添加左边的微信一起探讨!项目地址:(https://www.aliyun.com/activity/daily/bestoffer?userCodesskuuw5n)\2.Bug修复更新日历2.情侣脸功能大家不要使用了,现在阿里云的接口已经要收费了(土豪请随意),\\和注意
晴空闲云 晴空闲云
4个月前
css中box-sizing解放盒子实际宽高计算
我们知道传统的盒子模型,如果增加内边距padding和边框border,那么会撑大整个盒子,造成盒子的宽度不好计算,在实务中特别不方便。boxsizing可以设置盒模型的方式,可以很好的设置固定宽高的盒模型。盒子宽高计算假如我们设置如下盒子:宽度和高度均为200px,那么这会这个盒子实际的宽高就都是200px。但是当我们设置这个盒子的边框和内间距的时候,那
艾木酱 艾木酱
3个月前
快速入门|使用MemFire Cloud构建React Native应用程序
MemFireCloud是一款提供云数据库,用户可以创建云数据库,并对数据库进行管理,还可以对数据库进行备份操作。它还提供后端即服务,用户可以在1分钟内新建一个应用,使用自动生成的API和SDK,访问云数据库、对象存储、用户认证与授权等功能,可专
NVIDIA安培架构下MIG技术分析
关键词:NVIDIA、MIG、安培一什么是MIG2020年5月,NVIDIA发布了最新的GPU架构:安培,以及基于安培架构的最新的GPU:A100。安培提供了许多新的特性,MIG是其中一项非常重要的新特性。MIG的全名是MultiInstanceGPU。NVIDIA安培架构中的MIG模式可以在A100GPU上并行运行七个作业。多实
helloworld_28799839 helloworld_28799839
4个月前
常用知识整理
Javascript判断对象是否为空jsObject.keys(myObject).length0经常使用的三元运算我们经常遇到处理表格列状态字段如status的时候可以用到vue
密钥管理系统-为你的天翼云资产上把“锁
本文关键词:数据安全,密码机,密钥管理一、你的云上资产真的安全么?1.2021年1月,巴西的一个数据库30TB数据被破坏,泄露的数据包含有1.04亿辆汽车和约4000万家公司的详细信息,受影响的人员数量可能有2.2亿;2.2021年2月,广受欢迎的音频聊天室应用Clubhouse的用户数据被恶意黑客或间谍窃取。据悉,一位身份不明的用户能够将Clubho