1、 主流的四种限流策略，我都可以通过redis实现
引言
在web开发中功能是基石，除了功能以外运维和防护就是重头戏了。因为在网站运行期间可能会因为突然的访问量导致业务异常、也有可能遭受别人恶意攻击
所以我们的接口需要对流量进行限制。俗称的QPS也是对流量的一种描述
针对限流现在大多应该是令牌桶算法，因为它能保证更多的吞吐量。除了令牌桶算法还有他的前身漏桶算法和简单的计数算法
下面我们来看看这四种算法
2 固定时间窗口算法
2.1 说明
固定时间窗口算法也可以叫做简单计数算法。网上有很多都将计数算法单独抽离出来。但是笔者认为计数算法是一种思想，而固定时间窗口算法是他的一种实现
包括下面滑动时间窗口算法也是计数算法的一种实现。因为计数如果不和时间进行绑定的话那么失去了限流的本质了。就变成了拒绝了
优点
在固定的时间内出现流量溢出可以立即做出限流。每个时间窗口不会相互影响
在时间单元内保障系统的稳定。保障的时间单元内系统的吞吐量上限
缺点
正如图示一样，他的最大问题就是临界状态。在临界状态最坏情况会受到两倍流量请求
除了临界的情况，还有一种是在一个单元时间窗内前期如果很快地消耗完请求阈值。那么剩下的时间将会无法请求。这样就会因为一瞬间的流量导致一段时间内系统不可用。这在互联网高可用的系统中是不能接受的。
实现
好了，关于原理介绍及优缺点我们已经了解了。下面我们动手实现它
首先我们在实现这种计数时，采用redis是非常好的选择。这里我们通过redis实现
controller
@RequestMapping(value = "/start",method = RequestMethod.GET)
public Map<String,Object> start(@RequestParam Map<String, Object> paramMap) {
return testService.startQps(paramMap);
}
service
@Override
public Map<String, Object> startQps(Map<String, Object> paramMap) {
//根据前端传递的qps上线
Integer times = 100;
if (paramMap.containsKey(“times”)) {
times = Integer.valueOf(paramMap.get(“times”).toString());
}
String redisKey = “redisQps”;
RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(redisKey, redisTemplate.getConnectionFactory());
int no = redisAtomicInteger.getAndIncrement();
//设置时间固定时间窗口长度 1S
if (no == 0) {
redisAtomicInteger.expire(1, TimeUnit.SECONDS);
}
//判断是否超限 time=2 表示qps=3
if (no > times) {
throw new RuntimeException(“qps refuse request”);
}
//返回成功告知
Map<String, Object> map = new HashMap<>();
map.put(“success”, “success”);
return map;
}
2.2 lua实现原子性
pom.xml
<!-- spring boot redis缓存引入 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- 缓存连接池-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
<!-- redis 存储 json序列化 -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
配置一下redis配置信息
redis:
host: 127.0.0.1
port: 6379
database: 0
password: 123456 #默认为空
timeout: 3000ms #最大等待时间，超时则抛出异常，否则请求一直等待
lettuce:
pool:
max-active: 20 #最大连接数，负值表示没有限制，默认8
max-wait: -1 #最大阻塞等待时间，负值表示没限制，默认-1
max-idle: 8 #最大空闲连接，默认8
min-idle: 0 #最小空闲连接，默认0
整理一下思路
首先我们需要对接口做限流，那怎么限流呢？什么时候限流？
第一个问题怎么限流？
当我们接口被恶意调用的时候通常情况下有两种一种是我们对这个接口做全局限制，就是默认所有人在单位时间内只能调用多少次，第二种是对调用者的IP地址做限制
第二个问题什么时候限流？
当我们在执行这个方法之前就应该先判断是否限流，所以我们可以采用AOP+注解模式
所以我们现在就是要用注解+AOP对接口做两种模式限流：全局和IP地址
创建一个枚举类 来区别两种限流模式
public enum LimitType {
/**
* 默认策略全局限流
*/
DEFAULT,
/**
* 根据请求者IP进行限流
*/
IP
}
.我们创建一个注解，通过AOP去解析这个注解获取信息做出对应的判断
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
/**
* 限流key 这个key只是我们存入redis中的键值的前缀
*/
String key() default "rate_limit:";
/**
* 限流时间,单位秒
*/
int time() default 60;
/**
* 限流次数
*/
int count() default 100;
/**
* 限流类型
*/
LimitType limitType() default LimitType.DEFAULT;
}
配置RedisTemplate 创建一个RedisConfig 配置RedisTemplate （类上记得加@Configuration注解）
@Bean
public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(redisConnectionFactory);
//首先解决key的序列化方式
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
redisTemplate.setKeySerializer(stringRedisSerializer);
//解决value的序列化方式
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
//序列化时将类的数据类型存入json，以便反序列化的时候转换成正确的类型
ObjectMapper objectMapper = new ObjectMapper();
//objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);
// 解决jackson2无法反序列化LocalDateTime的问题
objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
objectMapper.registerModule(new JavaTimeModule());
jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
return redisTemplate;
}
接下来我们想一下怎么对redis操作的思路
首先当我们第一次调用这个接口的时候应该怎么做？
我们应该去先判断当前接口的具体参数(调用限制次数，限制时间)而这些东西都是应该存在redis里的，当我们对接口发起调用的时候，我们用一个设置唯一key去存并将它的值设置为自增，过期时间设定为限制时间。那么问题来了
我们应该如何去保证他们的操作原子性呢？
我们可以使用LUA脚本去实现这些操作,因为redis是单线程，所以在执行lua脚本的时候，会把脚本里面的所有命令当成一个命令执行，这样保证了redis的原子性。
创建一个LUA脚本放在resource目录下
–获取传递进来的KEY值
local key = KEYS[1]
–获取限制的次数
local count = tonumber(ARGV[1])
–获取限制的时间，设置为过期时间
local time = tonumber(ARGV[2])
–查询key是否存在并获取值 如果不存在返回nil
local current = redis.call(‘get’, key)
–如果key存在并且大于传入的限制次数 我们直接返回当前次数 不执行后续代码
if current and tonumber(current) > count then
return tonumber(current)
end
–走到这里代表次数没有超出限制 或者是第一次创建获取到的返回值为nil 所以我们给他自增1
current = redis.call(‘incr’, key)
–如果current==1 代表是第一次创建 因为redis.call(‘incr’, key)方法会对不存在key自动创建并赋值为0再自增返回1
if tonumber(current) == 1 then
–所以是第一次创建我们给他设置过期时间
redis.call(‘expire’, key, time)
end
–最后返回次数
return tonumber(current)
然后我们在一个Bean对象里面加载这个脚本
```javascript
```java
@Bean
public DefaultRedisScript<Long> limitScript() {
//执行 Redis Lua 脚本的一种方式
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
//将lua脚本加载ClassPathResource中
ClassPathResource resource = new ClassPathResource("lua/limit.lua");
//将脚本加载容器中方便后续执行
ResourceScriptSource resourceScriptSource = new ResourceScriptSource(resource);
redisScript.setScriptSource(resourceScriptSource);
redisScript.setResultType(Long.class);
return redisScript;
}
可以把这个Bean放在redis配置类中
注解解析 定义切面
创建一个
package com.maroon.Aspect;
import com.maroon.enums.LimitType;
import com.maroon.exception.BusinessException;
import com.maroon.interfaces.RateLimiter;
import com.maroon.util.IPUtil;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.List;
@Aspect
@Component
@Slf4j
public class RateLimiterAspect {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@Autowired
private RedisScript<Long> limitScript;
/**
* @param point
* @param rateLimiter
* @throws Throwable
*/
@Before("@annotation(rateLimiter)")
public void doBefore(JoinPoint point, RateLimiter rateLimiter) throws Throwable {
//从注解中获取 key time count
String key = rateLimiter.key();
int time = rateLimiter.time();
int count = rateLimiter.count();
//返回一个由 key：IP地址和切点方法的全路径加方法名的String
// 类似rate_limit:127.0.0.1-com.maroon.service.impl.SmsServiceImpl-sendCode
// （如果不是 IP 模式，那么生成的 key 中就不包含 IP 地址）
String combineKey = getCombineKey(rateLimiter, point);
//Collections.singletonList 是 Java 中的一个静态方法，用于创建一个只包含单个元素的不可变 List 集合。
// 此方法接受一个参数，即要放入 List 中的元素，然后返回一个不可变 List，包含指定的元素这个集合是不可修改的没有
List<String> keys = Collections.singletonList(combineKey);
Long number;
try {
Object[] o = {count, time};
//将键名和
number = redisTemplate.execute(limitScript, keys, o);
if (number == null || number.intValue() > count) {
throw new BusinessException("请求繁忙");
}
log.info("限制请求'{}',当前请求'{}',缓存key'{}'", count, number.intValue(), key);
} catch (BusinessException e) {
throw e;
} catch (Exception e) {
throw new RuntimeException("服务器限流异常，请稍候再试");
}
}
public String getCombineKey(RateLimiter rateLimiter, JoinPoint point) {
StringBuffer stringBuffer = new StringBuffer(rateLimiter.key());
if (rateLimiter.limitType() == LimitType.IP) {
//获取用户IP地址
stringBuffer.append(IPUtil.getIp()).append("-");
}
//获取被通知方法的签名信息 然后就可以获得该方法的名称、返回类型等信息。
MethodSignature signature = (MethodSignature) point.getSignature();
//获取切点方法对象
Method method = signature.getMethod();
//获取方法全路径名
Class<?> targetClass = method.getDeclaringClass();
//将IP地址和方法全路径名和方法名拼在一起返回
stringBuffer.append(targetClass.getName()).append("-").append(method.getName());
return stringBuffer.toString();
}
}
public class IPUtil {
public static String getIp() {
HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
String ip = "unknown";
if (request != null) {
ip = request.getHeader("X-Forwarded-For");
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getHeader("Proxy-Client-IP");
}
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getHeader("WL-Proxy-Client-IP");
}
if (ip == null || ip.length() == 0 || "unknown".equalsIgnoreCase(ip)) {
ip = request.getRemoteAddr();
}
}
return ip;
}
}
3 滑动时间窗口算法
针对固定时间窗口的缺点–临界值出现双倍流量问题。 我们的滑动时间窗口就产生了。
其实很好理解，就是针对固定时间窗口，将时间窗口统计从原来的固定间隔变成更加细度化的单元了。
在上面我们固定时间窗口演示中我们设置的时间单元是1S 。 针对1S我们将1S拆成时间戳。
固定时间窗口是统计单元随着时间的推移不断向后进行。而滑动时间窗口是我们认为的想象出一个时间单元按照相对论的思想将时间固定，我们的抽象时间单元自己移动。抽象的时间单元比实际的时间单元更小。
读者可以看下下面的动图，就可以理解了。
优点
实质上就是固定时间窗口算法的改进。所以固定时间窗口的缺点就是他的优点。
内部抽象一个滑动的时间窗，将时间更加小化。存在边界的问题更加小。客户感知更弱了。
缺点
不管是固定时间窗口算法还是滑动时间窗口算法，他们都是基于计数器算法进行优化，但是他们对待限流的策略太粗暴了。
为什么说粗暴呢，未限流他们正常放行。一旦达到限流后就会直接拒绝。这样我们会损失一部分请求。这对于一个产品来说不太友好
实现
滑动时间窗口是将时间更加细化，上面我们是通过redis#setnx实现的。这里我们就无法通过他统一记录了。我们应该加上更小的时间单元存储到一个集合汇总。然后根据集合的总量计算限流。redis的zsett数据结构就和符合我们的需求。
为什么选择zset呢，因为redis的zset中除了值以外还有一个权重。会根据这个权重进行排序。如果我们将我们的时间单元及时间戳作为我们的权重，那么我们获取统计的时候只需要按照一个时间戳范围就可以了。
因为zset内元素是唯一的，所以我们的值采用uuid或者雪花算法一类的id生成器
controller
@RequestMapping(value = "/startList",method = RequestMethod.GET)
public Map<String,Object> startList(@RequestParam Map<String, Object> paramMap) {
return testService.startList(paramMap);
}
service
String redisKey = "qpsZset";
Integer times = 100;
if (paramMap.containsKey("times")) {
times = Integer.valueOf(paramMap.get("times").toString());
}
long currentTimeMillis = System.currentTimeMillis();
long interMills = inter * 1000L;
Long count = redisTemplate.opsForZSet().count(redisKey, currentTimeMillis - interMills, currentTimeMillis);
if (count > times) {
throw new RuntimeException("qps refuse request");
}
redisTemplate.opsForZSet().add(redisKey, UUID.randomUUID().toString(), currentTimeMillis);
Map<String, Object> map = new HashMap<>();
map.put("success", "success");
return map;
和固定时间窗口采用相同的并发。为什么上面也会出现临界状况呢。因为在代码里时间单元间隔比固定时间间隔采用还要大 。 上面演示固定时间窗口时间单元是1S出现了最坏情况。而滑动时间窗口设计上就应该间隔更短。而我设置成10S 也没有出现坏的情况
这里就说明滑动比固定的优处了。如果我们调更小应该更加不会出现临界问题，不过说到底他还是避免不了临界出现的问题
4 漏桶算法
滑动时间窗口虽然可以极大程度地规避临界值问题，但是始终还是避免不了
另外时间算法还有个致命的问题，他无法面对突如其来的大量流量，因为他在达到限流后直接就拒绝了其他额外流量
针对这个问题我们继续优化我们的限流算法。 漏桶算法应运而生
优点
面对限流更加的柔性，不再粗暴的拒绝。
增加了接口的接收性
保证下流服务接收的稳定性。均匀下发
缺点
我觉得没有缺点。非要鸡蛋里挑骨头那我只能说漏桶容量是个短板
实现
controller
@RequestMapping(value = "/startLoutong",method = RequestMethod.GET)
public Map<String,Object> startLoutong(@RequestParam Map<String, Object> paramMap) {
return testService.startLoutong(paramMap);
}
service
在service中我们通过redis的list的功能模拟出桶的效果。这里代码是实验室性质的。在真实使用中我们还需要考虑并发的问题
@Override
public Map<String, Object> startLoutong(Map<String, Object> paramMap) {
String redisKey = "qpsList";
Integer times = 100;
if (paramMap.containsKey("times")) {
times = Integer.valueOf(paramMap.get("times").toString());
}
Long size = redisTemplate.opsForList().size(redisKey);
if (size >= times) {
throw new RuntimeException("qps refuse request");
}
Long aLong = redisTemplate.opsForList().rightPush(redisKey, paramMap);
if (aLong > times) {
//为了防止并发场景。这里添加完成之后也要验证。 即使这样本段代码在高并发也有问题。此处演示作用
redisTemplate.opsForList().trim(redisKey, 0, times-1);
throw new RuntimeException("qps refuse request");
}
Map<String, Object> map = new HashMap<>();
map.put("success", "success");
return map;
}
下游消费
@Component
public class SchedulerTask {
@Autowired
RedisTemplate redisTemplate;
private String redisKey="qpsList";
@Scheduled(cron="*/1 * * * * ?")
private void process(){
//一次性消费两个
System.out.println("正在消费。。。。。。");
redisTemplate.opsForList().trim(redisKey, 2, -1);
}
}
测试
我们还是通过50并发循环10次访问。我们可以发现只有在一开始能达到比较高的吞吐量。在随后桶的容量满了之后。而下游水滴速率比上游请求速率慢的情况下。只能以下游恒定的速度接收访问。
他的问题也暴露得很明显。针对时间窗口的不足漏桶进行的不足，但是仍是不足。无法彻底避免请求溢出的问题。
请求溢出本身就是一种灾难性的问题。所有的算法目前都没有解决这个问题。只是在减缓他带来的问题
5、令牌桶算法
令牌桶和漏桶法是一样的。只不过将桶的作用方向改变了一下。
漏桶的出水速度是恒定的，如果流量突然增加的话我们就只能拒绝入池
但是令牌桶是将令牌放入桶中，我们知道正常情况下令牌就是一串字符当桶满了就拒绝令牌的入池，但是面对高流量的时候正常加上我们的超时时间就留下足够长的时间生产及消费令牌了。这样就尽可能地不会造成请求的拒绝
最后，不论是对于令牌桶拿不到令牌被拒绝，还是漏桶的水满了溢出，都是为了保证大部分流量的正常使用，而牺牲掉了少部分流量
public Map<String, Object> startLingpaitong(Map<String, Object> paramMap) {
String redisKey = “lingpaitong”;
String token = redisTemplate.opsForList().leftPop(redisKey).toString();
//正常情况需要验证是否合法，防止篡改
if (StringUtils.isEmpty(token)) {
throw new RuntimeException(“令牌桶拒绝”);
}
Map<String, Object> map = new HashMap<>();
map.put(“success”, “success”);
return map;
}
@Scheduled(cron=“*/1 * * * * ?”)
private void process(){
//一次性生产两个
System.out.println(“正在消费。。。。。。”);
for (int i = 0; i < 2; i++) {
redisTemplate.opsForList().rightPush(redisKey, i);
}
}
![深度学习基础入门篇[9.1]：卷积之标准卷积：卷积核/特征图/卷积计算、填充、感受视野、多通道输入输出、卷积优势和应用案例讲解](https://img-blog.csdnimg.cn/img_convert/c70cf74e965c4cea8c139fb4632dcaff.png)
