java高并发系列 - 第15天：JUC中的Semaphore，最简单的限流工具类，必备技能

这是java高并发系列第15篇文章

Semaphore（信号量）为多线程协作提供了更为强大的控制方法，前面的文章中我们学了synchronized和重入锁ReentrantLock，这2种锁一次都只能允许一个线程访问一个资源，而信号量可以控制有多少个线程可以访问特定的资源。

Semaphore常用场景：限流

举个例子：

比如有个停车场，有5个空位，门口有个门卫，手中5把钥匙分别对应5个车位上面的锁，来一辆车，门卫会给司机一把钥匙，然后进去找到对应的车位停下来，出去的时候司机将钥匙归还给门卫。停车场生意比较好，同时来了100两车，门卫手中只有5把钥匙，同时只能放5辆车进入，其他车只能等待，等有人将钥匙归还给门卫之后，才能让其他车辆进入。

上面的例子中门卫就相当于Semaphore，车钥匙就相当于许可证，车就相当于线程。

Semaphore主要方法

  • Semaphore(int permits)：构造方法，参数表示许可证数量，用来创建信号量
  • Semaphore(int permits,boolean fair)：构造方法，当fair等于true时，创建具有给定许可数的计数信号量并设置为公平信号量
  • void acquire() throws InterruptedException：从此信号量获取1个许可前线程将一直阻塞，相当于一辆车占了一个车位，此方法会响应线程中断，表示调用线程的interrupt方法，会使该方法抛出InterruptedException异常
  • void acquire(int permits) throws InterruptedException ：和acquire()方法类似，参数表示需要获取许可的数量；比如一个大卡车要入停车场，由于车比较大，需要申请3个车位才可以停放
  • void acquireUninterruptibly(int permits) ：和acquire(int permits) 方法类似，只是不会响应线程中断
  • boolean tryAcquire()：尝试获取1个许可，不管是否能够获取成功，都立即返回，true表示获取成功，false表示获取失败
  • boolean tryAcquire(int permits)：和tryAcquire()，表示尝试获取permits个许可
  • boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException：尝试在指定的时间内获取1个许可，获取成功返回true，指定的时间过后还是无法获取许可，返回false
  • boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException：和tryAcquire(long timeout, TimeUnit unit)类似，多了一个permits参数，表示尝试获取permits个许可
  • void release()：释放一个许可，将其返回给信号量，相当于车从停车场出去时将钥匙归还给门卫
  • void release(int n)：释放n个许可
  • int availablePermits()：当前可用的许可数

示例1：Semaphore简单的使用

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：路人甲Java，专注于java技术分享（带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等），喜欢请关注！
 */
public class Demo1 {
    static Semaphore semaphore = new Semaphore(2);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            try {
                semaphore.acquire();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可!");
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",释放许可!");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            new T("t-" + i).start();
        }
    }
}

输出：

1563715791327,t-0,获取许可!
1563715791327,t-1,获取许可!
1563715794328,t-0,释放许可!
1563715794328,t-5,获取许可!
1563715794328,t-1,释放许可!
1563715794328,t-2,获取许可!
1563715797328,t-2,释放许可!
1563715797328,t-6,获取许可!
1563715797328,t-5,释放许可!
1563715797328,t-3,获取许可!
1563715800329,t-6,释放许可!
1563715800329,t-9,获取许可!
1563715800329,t-3,释放许可!
1563715800329,t-7,获取许可!
1563715803330,t-7,释放许可!
1563715803330,t-8,获取许可!
1563715803330,t-9,释放许可!
1563715803330,t-4,获取许可!
1563715806330,t-8,释放许可!
1563715806330,t-4,释放许可!

代码中new Semaphore(2)创建了许可数量为2的信号量，每个线程获取1个许可，同时允许两个线程获取许可，从输出中也可以看出，同时有两个线程可以获取许可，其他线程需要等待已获取许可的线程释放许可之后才能运行。为获取到许可的线程会阻塞在acquire()方法上，直到获取到许可才能继续。

示例2：获取许可之后不释放

门卫（Semaphore）有点呆，司机进去的时候给了钥匙，出来的时候不归还，门卫也不会说什么。最终结果就是其他车辆都无法进入了。

如下代码：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：路人甲Java，专注于java技术分享（带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等），喜欢请关注！
 */
public class Demo2 {
    static Semaphore semaphore = new Semaphore(2);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            try {
                semaphore.acquire();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可!");
                TimeUnit.SECONDS.sleep(3);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",运行结束!");
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",当前可用许可数量:" + semaphore.availablePermits());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            new T("t-" + i).start();
        }
    }
}

输出：

1563716603924,t-0,获取许可!
1563716603924,t-1,获取许可!
1563716606925,t-0,运行结束!
1563716606925,t-0,当前可用许可数量:0
1563716606925,t-1,运行结束!
1563716606925,t-1,当前可用许可数量:0

上面程序运行后一直无法结束，观察一下代码，代码中获取许可后，没有释放许可的代码，最终导致，可用许可数量为0，其他线程无法获取许可，会在semaphore.acquire();处等待，导致程序无法结束。

示例3：释放许可正确的姿势

示例1中，在finally里面释放锁，会有问题么？

如果获取锁的过程中发生异常，导致获取锁失败，最后finally里面也释放了许可，最终会怎么样，导致许可数量凭空增长了。

示例代码：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：路人甲Java，专注于java技术分享（带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等），喜欢请关注！
 */
public class Demo3 {
    static Semaphore semaphore = new Semaphore(1);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            try {
                semaphore.acquire();
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可,当前可用许可数量:" + semaphore.availablePermits());
                //休眠100秒
                TimeUnit.SECONDS.sleep(100);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",运行结束!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
            System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",当前可用许可数量:" + semaphore.availablePermits());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T t1 = new T("t1");
        t1.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t2 = new T("t2");
        t2.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t3 = new T("t3");
        t3.start();

        //给t2和t3发送中断信号
        t2.interrupt();
        t3.interrupt();
    }
}

输出：

1563717279058,t1,获取许可,当前可用许可数量:0
java.lang.InterruptedException
1563717281060,t2,当前可用许可数量:1
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
1563717281060,t3,当前可用许可数量:2
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo3$T.run(Demo3.java:21)
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo3$T.run(Demo3.java:21)

程序中信号量许可数量为1，创建了3个线程获取许可，线程t1获取成功了，然后休眠100秒。其他两个线程阻塞在semaphore.acquire();方法处，代码中对线程t2、t3发送中断信号，我们看一下Semaphore中acquire的源码：

public void acquire() throws InterruptedException

这个方法会响应线程中断，主线程中对t2、t3发送中断信号之后，acquire()方法会触发InterruptedException异常，t2、t3最终没有获取到许可，但是他们都执行了finally中的释放许可的操作，最后导致许可数量变为了2，导致许可数量增加了。所以程序中释放许可的方式有问题。需要改进一下，获取许可成功才去释放锁。

正确的释放锁的方式，如下：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：路人甲Java，专注于java技术分享（带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等），喜欢请关注！
 */
public class Demo4 {
    static Semaphore semaphore = new Semaphore(1);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            //获取许可是否成功
            boolean acquireSuccess = false;
            try {
                semaphore.acquire();
                acquireSuccess = true;
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可,当前可用许可数量:" + semaphore.availablePermits());
                //休眠100秒
                TimeUnit.SECONDS.sleep(5);
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",运行结束!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (acquireSuccess) {
                    semaphore.release();
                }
            }
            System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",当前可用许可数量:" + semaphore.availablePermits());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T t1 = new T("t1");
        t1.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t2 = new T("t2");
        t2.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t3 = new T("t3");
        t3.start();

        //给t2和t3发送中断信号
        t2.interrupt();
        t3.interrupt();
    }
}

输出：

1563717751655,t1,获取许可,当前可用许可数量:0
1563717753657,t3,当前可用许可数量:0
java.lang.InterruptedException
1563717753657,t2,当前可用许可数量:0
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1302)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo4$T.run(Demo4.java:23)
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
    at java.util.concurrent.Semaphore.acquire(Semaphore.java:312)
    at com.itsoku.chat12.Demo4$T.run(Demo4.java:23)
1563717756656,t1,运行结束!
1563717756656,t1,当前可用许可数量:1

程序中增加了一个变量acquireSuccess用来标记获取许可是否成功，在finally中根据这个变量是否为true，来确定是否释放许可。

示例4：在规定的时间内希望获取许可

司机来到停车场，发现停车场已经满了，只能在外等待内部的车出来之后才能进去，但是要等多久，他自己也不知道，他希望等10分钟，如果还是无法进去，就不到这里停车了。

Semaphore内部2个方法可以提供超时获取许可的功能：

public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException

在指定的时间内去尝试获取许可，如果能够获取到，返回true，获取不到返回false。

示例代码：

package com.itsoku.chat12;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：路人甲Java，专注于java技术分享（带你玩转 爬虫、分布式事务、异步消息服务、任务调度、分库分表、大数据等），喜欢请关注！
 */
public class Demo5 {
    static Semaphore semaphore = new Semaphore(1);

    public static class T extends Thread {
        public T(String name) {
            super(name);
        }

        @Override
        public void run() {
            Thread thread = Thread.currentThread();
            //获取许可是否成功
            boolean acquireSuccess = false;
            try {
                //尝试在1秒内获取许可，获取成功返回true，否则返回false
                System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",尝试获取许可,当前可用许可数量:" + semaphore.availablePermits());
                acquireSuccess = semaphore.tryAcquire(1, TimeUnit.SECONDS);
                //获取成功执行业务代码
                if (acquireSuccess) {
                    System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可成功,当前可用许可数量:" + semaphore.availablePermits());
                    //休眠5秒
                    TimeUnit.SECONDS.sleep(5);
                } else {
                    System.out.println(System.currentTimeMillis() + "," + thread.getName() + ",获取许可失败,当前可用许可数量:" + semaphore.availablePermits());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                if (acquireSuccess) {
                    semaphore.release();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T t1 = new T("t1");
        t1.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t2 = new T("t2");
        t2.start();
        //休眠1秒
        TimeUnit.SECONDS.sleep(1);
        T t3 = new T("t3");
        t3.start();
    }
}

输出：

1563718410202,t1,尝试获取许可,当前可用许可数量:1
1563718410202,t1,获取许可成功,当前可用许可数量:0
1563718411203,t2,尝试获取许可,当前可用许可数量:0
1563718412203,t3,尝试获取许可,当前可用许可数量:0
1563718412204,t2,获取许可失败,当前可用许可数量:0
1563718413204,t3,获取许可失败,当前可用许可数量:0

代码中许可数量为1，semaphore.tryAcquire(1, TimeUnit.SECONDS);：表示尝试在1秒内获取许可，获取成功立即返回true，超过1秒还是获取不到，返回false。线程t1获取许可成功，之后休眠了5秒，从输出中可以看出t2和t3都尝试了1秒，获取失败。

其他一些使用说明

  1. Semaphore默认创建的是非公平的信号量，什么意思呢？这个涉及到公平与非公平。举个例子：5个车位，允许5个车辆进去，来了100辆车，只能进去5辆，其他95在外面排队等着。里面刚好出来了1辆，此时刚好又来了10辆车，这10辆车是直接插队到其他95辆前面去，还是到95辆后面去排队呢？排队就表示公平，直接去插队争抢第一个，就表示不公平。对于停车场，排队肯定更好一些咯。不过对于信号量来说不公平的效率更高一些，所以默认是不公平的。
  2. 建议阅读以下Semaphore的源码，对常用的方法有个了解，不需要都记住，用的时候也方便查询就好。
  3. 方法中带有throws InterruptedException声明的，表示这个方法会响应线程中断信号，什么意思？表示调用线程的interrupt()方法，会让这些方法触发InterruptedException异常，即使这些方法处于阻塞状态，也会立即返回，并抛出InterruptedException异常，线程中断信号也会被清除。

java高并发系列

java高并发系列连载中，总计估计会有四五十篇文章，可以关注公众号：javacode2018，获取最新文章。

原文地址：https://www.cnblogs.com/itsoku123/p/11223837.html

时间： 2024-07-30 10:14:14

java高并发系列 - 第15天：JUC中的Semaphore，最简单的限流工具类，必备技能的相关文章

java高并发系列 - 第32天：高并发中计数器的实现方式有哪些？

这是java高并发系列第32篇文章. java环境:jdk1.8. 本文主要内容 4种方式实现计数器功能,对比其性能 介绍LongAdder 介绍LongAccumulator 需求:一个jvm中实现一个计数器功能,需保证多线程情况下数据正确性. 我们来模拟50个线程,每个线程对计数器递增100万次,最终结果应该是5000万. 我们使用4种方式实现,看一下其性能,然后引出为什么需要使用LongAdder.LongAccumulator. 方式一:synchronized方式实现 package

java限流工具类

代码 import com.google.common.util.concurrent.RateLimiter; import java.util.concurrent.ConcurrentHashMap; /** * 限流工具类 * @author ZhangShuzheng * @date 2018/8/13 */ public class RateLimiterManager { private static ConcurrentHashMap<String, RateLimiter>

Java高并发系列---第1天(概念)

同步(Synchronous)和异步(Asynchronous) 同步和异步通常来形容一次方法调用,同步方法调用一旦开始,调用者必须等到方法调用返回后,才能继续后续的行为. 异步方法调用更像一个消息传递,一旦开始,方法调用就会立即返回,调用者就可以继续后续的操作. 而异步方法通常会在另外一个线程中“真实”地执行.整个过程,不会阻碍调用者的工作. 上图中显示了同步方法调用和异步方法调用的区别.对于调用者来说,异步调用似乎是一瞬间就完成的.如果异步调用需要返回结果,那么当这个异步调用真实完成时,则会

跟着阿里p7一起学java高并发 - 第18天：玩转java线程池，这一篇就够了

java中的线程池,这一篇就够了 java高并发系列第18篇文章. 本文主要内容 什么是线程池 线程池实现原理 线程池中常见的各种队列 自定义线程创建的工厂 常见的饱和策略 自定义饱和策略 线程池中两种关闭方法有何不同 扩展线程池 合理地配置线程池 线程池中线程数量的配置 什么是线程池 大家用jdbc操作过数据库应该知道,操作数据库需要和数据库建立连接,拿到连接之后才能操作数据库,用完之后销毁.数据库连接的创建和销毁其实是比较耗时的,真正和业务相关的操作耗时是比较短的.每个数据库操作之前都需要创

Java高并发秒杀API之业务分析与DAO层

课程介绍 高并发和秒杀都是当今的热门词汇,如何使用Java框架实现高并发秒杀API是该系列课程要研究的内容.秒杀系列课程分为四门,本门课程是第一门,主要对秒杀业务进行分析设计,以及DAO层的实现.课程中使用了流行的框架组合SpringMVC+Spring+MyBatis,还等什么,赶快来加入吧! 第1章 课程介绍 本章介绍秒杀系统的技术内容,以及系统演示.并介绍不同程度的学员可以学到什么内容. 第2章 梳理所有技术和搭建工程 本章首先介绍秒杀系统所用框架和技术点,然后介绍如何基于maven搭建项

2017.4.26 慕课网--Java 高并发秒杀API（一）

Java高并发秒杀API系列(一) -----------------业务分析及Dao层 第一章 课程介绍 1.1 内容介绍及业务分析 (1)课程内容 1 SSM框架的整合使用 2 秒杀类系统需求理解和实现 3 常用技术解决高并发问题 (2)SSM框架 (3)为何选择秒杀系统 1 秒杀系统具有典型的"事务"特性 2 秒杀/红包类需求越来越常见 3 面试常用问题 1.3 项目效果演示 第二章 梳理所有技术和搭建工程 2.1 相关技术介绍 2.2 创建项目和依赖 第三章 秒杀业务分析 3.

【实战Java高并发程序设计 4】数组也能无锁：AtomicIntegerArray

除了提供基本数据类型外,JDK还为我们准备了数组等复合结构.当前可用的原子数组有:AtomicIntegerArray.AtomicLongArray和AtomicReferenceArray,分别表示整数数组.long型数组和普通的对象数组. 这里以AtomicIntegerArray为例,展示原子数组的使用方式. AtomicIntegerArray本质上是对int[]类型的封装.使用Unsafe类通过CAS的方式控制int[]在多线程下的安全性.它提供了以下几个核心API: //获得数组第

构建高性能服务（二）java高并发锁的3种实现

构建高性能服务(二)java高并发锁的3种实现 来源:http://www.xymyeah.com/?p=46 提高系统并发吞吐能力是构建高性能服务的重点和难点.通常review代码时看到synchronized是我都会想一想,这个地方可不可以优化.使用synchronized使得并发的线程变成顺序执行,对系统并发吞吐能力有极大影响,我的博文 http://maoyidao.iteye.com/blog/1149015 介绍了可以从理论上估算系统并发处理能力的方法. 那么对于必须使用synchr

【实战Java高并发程序设计 3】带有时间戳的对象引用：AtomicStampedReference

[实战Java高并发程序设计 1]Java中的指针:Unsafe类 [实战Java高并发程序设计 2]无锁的对象引用:AtomicReference AtomicReference无法解决上述问题的根本是因为对象在修改过程中,丢失了状态信息.对象值本身与状态被画上了等号.因此,我们只要能够记录对象在修改过程中的状态值,就可以很好的解决对象被反复修改导致线程无法正确判断对象状态的问题. AtomicStampedReference正是这么做的.它内部不仅维护了对象值,还维护了一个时间戳(我这里把它