高级JAVA - 高并发下接口限流 Semaphore

Semaphore的介绍

Semaphore, 是JDK1.5的java.util.concurrent并发包中提供的一个并发工具类

Semaphore字面意思即信号量 , 个人认为比较容易理解的说法应该是许可证管理器

官方的解释为

Semaphore是一个计数信号量
从概念上将,Semaphore包含一组许可证
如果有需要的话,每次调用acquire()方法都会阻塞,直到获取一个可用的许可证
每次调用release()方法都会释放持有许可证的线程,并且归还Semaphore一个可用的许可证
实际上并没有真实的许可证对象供线程使用, Semaphore只是对可用的数量进行管理维护

Semaphore内部主要通过AQS(AbstractQueuedSynchronizer)实现线程的管理 .

线程在运行时首先获取许可 , 如果成功 , 许可数就减1 , 线程运行 ; 当线程运行结束就释放许可 , 许可数就加1 ; 如果许可数为0 , 则获取失败 , 线程位于AQS的等待队列中 , 它会被其它释放许可的线程唤醒

Semaphore的构造器

Semaphore(int permits) : 初始化指定数量许可证的Semaphore
Semaphore(int permits, boolean fair) : 初始化指定数量许可证的Semaphore , 并指定是否公平模式(即FIFO先进先出原则)

Semaphore的常用的一些方法

boolean isFail() : 当前Semaphore是否是公平模式
int availablePermits() : 当前Semaphore可用的许可证数量
boolean hasQueuedThreads() : 判断当前Semaphore中是否存在正在等待许可证的线程
int getQueueLength() : 获取当前Semaphore中正在等待许可证的线程数量void acquire() throws InterruptedException : 尝试获取一个可用的许可证 , 如果无可用许可证 , 当前线程会阻塞 , 如果线程被中断则抛出InterruptedException , 并停止阻塞继续执行
void acquire(int permits) throws InterruptedException : 尝试获取指定数量的可用许可证 , 如果无可用或数量不足 , 当前线程会阻塞 ,
如果线程被中断则抛出InterruptedException , 并停止阻塞继续执行
boolean tryAcquire() : 尝试获取1次可用许可证 , 非阻塞的 , 仅在调用该方法时尝试一次 , 获取到就继续执行并返回true . 获取不到也会停止等待继续执行 , 并返回false . 此外有重载方法tryAcquire(int permits)
boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException : 在限定时间内尝试获取1个可用许可证 , 如果获取到则继续执行返回true , 如果超时则继续执行并返回false,如果被中断则抛出一次异常并继续执行,此外有重载方法tryAcquire(int permits, long timeout, TimeUnit unit)
void acquireUninterruptibly() : 不可中断的尝试获取1个可用许可证 , 此外有重载方法acquireUninterruptibly(int permits)
int drainPermits() : 获取剩余所有可用的许可证void release() : 释放1个可用的许可证 , 重载方法acquire(int permits)

Semaphore的代码演示

package com.xing.apiLimiting;import lombok.AllArgsConstructor;
import org.apache.commons.lang3.RandomUtils;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;/*** @title 信号量Semaphore* @author Xingbz* @createDate 2018-12-7*/
public class SemaphoreDemo {public static void main(String[] args) throws Exception {
//        demo1();//Semaphore的初始化 / 基本属性 / 获取释放操作
//        demo2();//多进程之间的获取/释放/等待
//        demo3();//实例场景}/*** @title Semaphore的初始化 / 基本属性 / 获取释放操作* @author Xingbz*/private static void demo1() throws Exception {Semaphore semaphore1 = new Semaphore(5);//初始化指定数量许可证的SemaphoreSemaphore semaphore2 = new Semaphore(5, true);//初始化指定数量许可证的Semaphore , 并指定为公平模式(FIFO)System.out.println("是否公平模式 : semaphore1 " + semaphore1.isFair() + "; semaphore2 " + semaphore2.isFair());//是否公平模式FIFOsemaphore2.acquire();//尝试获取一个可用的许可证System.out.println("成功获取1个可用许可证 ; 剩余可用许可证数量 : " + semaphore2.availablePermits());semaphore2.release();//释放1个许可证System.out.println("成功释放1个许可证 ; 剩余可用许可证数量 : " + semaphore2.availablePermits());semaphore2.acquire(2);//尝试获取N个可用的许可证System.out.println("成功获取2个可用许可证 ; 剩余可用许可证数量 : " + semaphore2.availablePermits());semaphore2.release(2);//释放N个许可证 // 此时如果释放3个(>2) , 则可用数量亦会增加为6个?System.out.println("成功释放2个许可证 ; 剩余可用许可证数量 : " + semaphore2.availablePermits() + "\n");}/*** @title 多进程之间的获取/释放/等待* @author Xingbz*/private static void demo2() throws Exception {Semaphore semaphore = new Semaphore(5, true);new Thread(() -> {//启动一个新线程获取所有可用许可证System.out.println(Thread.currentThread().getName() + "尝试获取剩余所有可用的许可证 . . .");int permits = semaphore.drainPermits();//获取剩余所有可用的许可证System.out.println(Thread.currentThread().getName() + "成功获取" + permits + "个可用许可证 ; 剩余可用许可证数量 : " + semaphore.availablePermits());try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}semaphore.release(permits);System.out.println(Thread.currentThread().getName() + "成功释放" + permits + "个许可证 ; 剩余可用许可证数量 : " + semaphore.availablePermits());}).start();Thread.sleep(100);System.out.println("\n[1]当前剩余可用的许可证数量 : " + semaphore.availablePermits());System.out.println("[1]否存在正在等待许可证的线程 : " + semaphore.hasQueuedThreads());System.out.println("[1]正在等待许可证的线程数量 : " + semaphore.getQueueLength());new Thread(() -> {//启动另外一个新线程尝试获取1个许可证try {System.out.println(Thread.currentThread().getName() + "尝试获取1个可用许可证 . . .");semaphore.acquire();//由于前一个线程获取了所有可用 , 所以这里会阻塞 , 直到前一个线程释放1个为止System.out.println(Thread.currentThread().getName() + "成功获取1个可用许可证 ; 剩余可用许可证数量 : " + semaphore.availablePermits());} catch (InterruptedException e) {e.printStackTrace();}semaphore.release();System.out.println(Thread.currentThread().getName() + "成功释放1个许可证 ; 剩余可用许可证数量 : " + semaphore.availablePermits());}).start();Thread.sleep(100);System.out.println("\n[2]当前剩余可用的许可证数量 : " + semaphore.availablePermits());System.out.println("[2]否存在正在等待许可证的线程 : " + semaphore.hasQueuedThreads());System.out.println("[2]正在等待许可证的线程数量 : " + semaphore.getQueueLength());new Thread(() -> {//启动另外一个新线程尝试获取2个许可证try {System.out.println(Thread.currentThread().getName() + "尝试获取2个可用许可证 . . .");semaphore.acquire(2);System.out.println(Thread.currentThread().getName() + "成功获取2个可用许可证 ; 剩余可用许可证数量 : " + semaphore.availablePermits());} catch (InterruptedException e) {e.printStackTrace();}semaphore.release(2);System.out.println(Thread.currentThread().getName() + "成功释放2个许可证 ; 剩余可用许可证数量 : " + semaphore.availablePermits());}).start();Thread.sleep(100);System.out.println("\n[3]当前剩余可用的许可证数量 : " + semaphore.availablePermits());System.out.println("[3]否存在正在等待许可证的线程 : " + semaphore.hasQueuedThreads());System.out.println("[3]正在等待许可证的线程数量 : " + semaphore.getQueueLength());Thread.sleep(6000);System.out.println("\n[终]当前剩余可用的许可证数量 : " + semaphore.availablePermits());System.out.println("[终]否存在正在等待许可证的线程 : " + semaphore.hasQueuedThreads());System.out.println("[终]正在等待许可证的线程数量 : " + semaphore.getQueueLength());}private static final ExecutorService executorService = Executors.newCachedThreadPool();/*** @title 实例场景* @description*  场景说明 :*  1 . 模拟出口车辆许可的场景*  2 . 路段有2个车辆出口*          初始化一个permits=2的Semaphore对象*  3 . 某时段内共有20个车辆依次出行*          该Semaphore对象遵循公平(FIFO)原则 , 定义20个请求线程*  4 . 每个车辆出去的速度不同 , 耗费时间也不同*  5 . 有的车辆会一直等待到出口放行*          调用acquireUninterruptibly()方法 , 不会中断*  6 . 有的车辆没有耐心 , 等待时间超过自己预期 , 则不再排队 , 折返回去*          调用tryAcquire(timeout,TimeUnit)方法 , 且等待超时*  7 . 有的车辆很有耐心 , 但突然接到电话 , 不再排队 , 去其他地点了*          调用了acquire()方法 , 且被中断* @author Xingbz*/private static void demo3() throws Exception {Semaphore semaphore = new Semaphore(2, true);List<Future> futureList = new ArrayList<>(5);//某车队的车辆 , 排队中途会被电话中断for (int i = 1; i <= 20; i++) {if (i <= 10) {//前10辆车耐心排队等候许可executorService.execute(new Car("耐心车辆" + i, semaphore, 1));} else if (i <= 15) {//这5个车辆没有耐心  , 只愿等待少许executorService.execute(new Car("折返车辆" + i, semaphore, 2));} else {//这5个车辆被中断Future future = executorService.submit(new Thread(new Car("中断车辆" + i, semaphore, 3)));futureList.add(future);}}Thread.sleep(10000);//整体排队10S后 , 某车队被电话集体召回System.out.println("车队众人接到电话 , 集体终止排队");futureList.forEach(f -> f.cancel(true));}@AllArgsConstructorstatic class Car implements Runnable {/** 车辆编号 */private String carNo;/** 许可证 */private Semaphore semaphore;/** 行动方式 : 1一直等待直到许可 ; 2没有耐心折返回去 ; 3等待中被中断 */private int type;/*** @title 许可* @author Xingbz*/@Overridepublic void run() {switch (type) {case 1: //该车辆很有耐心  ,一直等到许可semaphore.acquireUninterruptibly();try {Thread.sleep(RandomUtils.nextLong(3000, 4000));//正在许可 . . .} catch (InterruptedException e) {}semaphore.release();//许可完毕 , 将许可证归还System.out.println(carNo + "已成功许可 . ");break;case 2://该车辆没有耐心 , 等待1s没有获取许可证 , 就折返了try {if (semaphore.tryAcquire(RandomUtils.nextInt(6000, 11000), TimeUnit.MILLISECONDS)) {Thread.sleep(RandomUtils.nextLong(3000, 4000));//正在许可 . . .semaphore.release();//许可完毕 , 将许可证归还System.out.println(carNo + "已成功许可 . ");} else {//超时则不再等待System.out.println(carNo + "折返 . ");}} catch (InterruptedException e) {}break;case 3://该车辆等待中被中断try {semaphore.acquire();Thread.sleep(RandomUtils.nextLong(3000, 4000));//正在许可 . . .semaphore.release();System.out.println(carNo + "已成功许可 . ");} catch (InterruptedException e) {//被电话 中断  , 不再等待System.out.println(carNo + "中断 , 驶向他处 . ");}}}}
}

高级JAVA - 高并发下接口限流 Semaphore相关推荐

高并发处理之接口限流
最近开发的抢购活动上线后发现了两个比较明显的问题,其一:活动一开始,接口访问量剧增:其二:黑名单中增加了一大批黑名单用户(或者说IP),这其中就包含了一些恶意用户或机器人刷接口. 针对一些高并发的接口 ...
Java高并发系统的限流策略
限流算法令牌桶(Token Bucket).漏桶(leaky bucket)和计数器算法是最常用的三种限流的算法. 计数器限流算法也是比较常用的,主要用来限制总并发数,比如数据库连接池大小.线程池大 ...
java guava限流,Guava的RateLimiter实现接口限流
最近开发需求中有需要对后台接口进行限流处理,整理了一下基本使用方法. 首先添加guava依赖: com.google.guava guava 23.0 然后封装RateLimiter适用对多接口的限制 ...
Java接口限流算法
0. 前言常见的限流算法有:令牌桶.漏桶.计数器也可以进行粗暴限流实现. 1. 算法介绍 1.1 令牌桶算法令牌桶算法是一个存放固定容量令牌的桶,按照固定速率往桶里添加令牌.令牌桶算法的描述如下: ...
java 接口防刷_java轻量级接口限流/防刷插件
简介 call-limit提供接口限流.防刷的功能,插件基于spring开发,在应用应用的任何一个逻辑层皆可使用(web.service.dao), 插件支持单机应用下的限流和分布式应用的限流(分布式 ...
高并发中的限流、熔断、降级、预热、背压你都知道是什么意思吗？
首先,我们需要明确一下这几个名词出现的场景:分布式高并发环境.如果你的产品卖相不好,没人鸟它,那它就用不着这几个属性.不需要任何加成,低并发系统就能工作的很好. 分布式系统是一个整体,调用关系错综复杂 ...
高并发系统之限流特技
转载至:http://blog.csdn.net/g_hongjin/article/details/51649246 在开发高并发系统时有三把利器用来保护系统:缓存.降级和限流.缓存的目的是提升系统 ...
接口限流算法及解决方案
参考: 接口限流算法:漏桶算法&令牌桶算法文章目录一.限流算法 1. 漏桶算法 2. 令牌桶算法二.令牌桶算法VS漏桶算法三.解决方案 1. 使用Guava的RateLimiter进行 ...
聊聊高并发系统之限流特技-1
在开发高并发系统时有三把利器用来保护系统:缓存.降级和限流. 缓存的目的是提升系统访问速度和增大系统能处理的容量,可谓是抗高并发流量的银弹:而降级是当服务出问题或者影响到核心流程的性能则需要暂时屏蔽掉 ...