一、限流前言
1.1、网关限流1.2、服务端限流--单机限流1.3、服务端限流--分布式限流1.4、实际应用---RateLimiter 二、分布式限流--redis lua+令牌桶算法
2.1、springboot集成redisTemplate2.2、redis lua+令牌桶算法代码实现
一、限流前言无论是传统还是互联网技术框架，我们要保证系统的高可用----减少系统不能提供服务的时间，保证服务高度可用性。其中，保障高可用的实践方案之一：限流。
按照个人理解，限流可分为如下：
一般是借用第三方组件技术来实现，主要在高并发场景的系统架构使用的较多。
1.2、服务端限流–单机限流主要有三种常见的算法：计数器、漏桶算法、令牌桶算法
计数器
如限流QPS为100，那么1秒内最多100个请求，超过的给拒绝。
漏桶算法
一个固定容量的漏桶按照常量固定速率流出水滴。能够让突发流量被整形，以便为网络提供稳定的流量。
流入桶的速率任意，如果桶容量超过了，那么就丢弃水滴。但流出桶的速率固定不变【无法解决系统突发流量】。
令牌桶算法
保证均匀速度产生令牌，一个令牌对应处理一个请求。同时，产生的令牌数量达到了桶的容量时，会被丢弃。那么能很好地处理突发高流量。
令牌桶有一定容量，后台服务以恒定速率向桶中放入令牌(token)，当桶中数量超过容量后，多余的令牌直接丢弃。当N个请求进入，从桶中拿N个令牌，能拿到的请求继续后续流程；拿不到的请求选择阻塞等待或返回失败。
在分布式系统中，有成熟的组件来帮助实现分布式限流：如Sentinel
1.4、实际应用—RateLimiterRateLimiter类是Guava提供的，可以限制单台服务器的接口流量。
即使在分布式系统中，也可以粗糙的使用。预估整个系统的流量限制all_limit，拥有n个服务器，平均每个服务器限流大小是 all_limit/n。【这种情况比较理想预估，不严谨】
考虑实际项目的技术选型，引用第三方技术栈会带来维护难度，所以，这里借助“redis lua+令牌桶算法思想”来实现分布式限流。
2.1、springboot集成redisTemplate引用jar包
org.springframework.boot
spring-boot-starter-data-redis
配置参数
#Jedis版本是直接连接redis server，Lettuce版本是基于Netty，保证多线程并发 spring.redis.host=127.0.0.1 spring.redis.port=6379 spring.redis.password= spring.redis.database=0 #最大活跃链接数 spring.redis.lettuce.pool.max-active=10 #最大空闲链接数 spring.redis.lettuce.pool.max-idle=10 #最小活跃链接数 spring.redis.lettuce.pool.min-active=0 #最小空闲链接数 spring.redis.lettuce.pool.min-idle=5
componennt注入redisTemplate
@Configuration
public class RedisLettuceConfig {
@Value("${spring.redis.lettuce.pool.max-idle}")
private int maxIdle;
@Value("${spring.redis.lettuce.pool.min-idle}")
private int minIdle;
@Value("${spring.redis.lettuce.pool.max-active}")
private int maxActive;
@Bean("redisTemplate")
public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory){
RedisTemplate template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
//序列化value值
Jackson2JsonRedisSerializer serializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper mapper = new ObjectMapper();
mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
serializer.setObjectMapper(mapper);
template.setValueSerializer(serializer);
//使用序列化Key值
template.setKeySerializer(new StringRedisSerializer());
template.setHashKeySerializer(new StringRedisSerializer());
template.setHashValueSerializer(serializer);
template.afterPropertiesSet();
return template;
}
}
2.2、redis lua+令牌桶算法代码实现
获取token的代码
@Component
public class RedisRateLimitClient {
@Autowired
@Qualifier("redisTemplate")
private RedisTemplate redisTemplate;
private DefaultRedisscript ratelimitInitLua;
@PostConstruct
public void init() {
ratelimitInitLua = new DefaultRedisscript();
ratelimitInitLua.setResultType(Number.class);
ratelimitInitLua.setscriptSource(new ResourcescriptSource(new ClassPathResource("redis/rateLimiter.lua")));
}
public boolean accquireToken(String key, Integer permits,String origin) {
///markSet 定义标记，比如哪些是被限流的，这个可以在哪配置
Set markSet = new HashSet<>();
markSet.add("mark-net");
markSet.add("mark-1");
if(!markSet.contains(origin)){
return false; //表示未配置，不限制流
}
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new StringRedisSerializer());
Long currMillSecond = redisTemplate.execute(
(RedisCallback) redisConnection -> redisConnection.time()
);
Integer min_permits = 3;//令牌桶的最少令牌数
Integer max_permits = 1000;//令牌桶的最大令牌数
Integer rate = 100;//向令牌桶中添加令牌的速率 , 令牌消耗速率【速率 100个/s】
Number accquire = redisTemplate.execute
(ratelimitInitLua,
Collections.singletonList("ratelimit:"+key),
currMillSecond,
min_permits,
max_permits,
rate,
origin,
permits);
boolean token;
if (accquire != null && accquire.intValue() == 1) {
token = true;
} else {
token = false;
}
return token;
}
}
redis/rateLimiter.lua实现
--先从redis缓存读取上一次操作读取令牌的情况
local ratelimit_info=redis.pcall("HMGET",KEYS[1],"last_mill_second","min_permits","max_permits","rate","mark");
local last_mill_second= ratelimit_info[1];
local min_permits= ratelimit_info[2];
local max_permits=ratelimit_info[3];
local rate= ratelimit_info[4];
local mark= ratelimit_info[5];
-- ARGV[1] : 当前操作的时间戳
-- ARGV[2] : 最小令牌数
-- ARGV[3] : 最大令牌数
-- ARGV[4] : 每秒速率
-- ARGV[5] : 标识哪个场景
-- ARGV[6] : 此次操作需要消耗令牌数
if mark == nil then
--先进行初始化
redis.pcall("HMSET",KEYS[1],"last_mill_second",ARGV[1],
"min_permits",ARGV[2],"max_permits",ARGV[3],"rate",ARGV[4],"mark",ARGV[5]);
last_mill_second = ARGV[1];
min_permits = ARGV[2];
max_permits=ARGV[3];
rate=ARGV[4];
mark=ARGV[5];
end
--计算时间戳范围内 应该产生多少令牌数
local reverse_permits=math.floor(tonumber(ARGV[1]-last_mill_second)/1000)*rate;
if reverse_permits>0 then
redis.pcall("HMSET",KEYS[1],"last_mill_second",ARGV[1]);
end
--这是计算出到达当前时刻，应该产生多少令牌数了
local expect_curr_permits=reverse_permits+min_permits;
local local_curr_permits=math.min(expect_curr_permits,max_permits);
local result=-1;
local rev_permit =local_curr_permits-ARGV[6];
if rev_permit>0 then
-- 此次消耗的令牌数够了
result=1;
redis.pcall("HMSET",KEYS[1],"min_permits",rev_permit);
else
redis.pcall("HMSET",KEYS[1],"min_permits",local_curr_permits);
end
return result;
我们是采用Redis的Hash类型数据结构来实现。
* last_mill_second【上一次消耗令牌桶时间戳】
* min_permits【最小令牌数】
* max_permits【最大令牌数】
* rate【每秒生成几个令牌】
* mark【匹配上，才生效】
redis lua保证了原子性，通过这种方式可以简单实现分布式限流。
