Java并发：分布式应用限流 Redis + Lua 实践

猫猫1

　　任何限流都不是漫无目的的，也不是一个开关就可以解决的问题，常用的限流算法有：令牌桶，漏桶。在之前的文章中，也讲到过，但是那是基于单机场景来写。
　　之前文章：接口限流算法：漏桶算法&令牌桶算法
　　然而再牛逼的机器，再优化的设计，对于特殊场景我们也是要特殊处理的。就拿秒杀来说，可能会有百万级别的用户进行抢购，而商品数量远远小于用户数量。如果这些请求都进入队列或者查询缓存，对于最终结果没有任何意义，徒增后台华丽的数据。对此，为了减少资源浪费，减轻后端压力，我们还需要对秒杀进行限流，只需保障部分用户服务正常即可。
　　就秒杀接口来说，当访问频率或者并发请求超过其承受范围的时候，这时候我们就要考虑限流来保证接口的可用性，以防止非预期的请求对系统压力过大而引起的系统瘫痪。通常的策略就是拒绝多余的访问，或者让多余的访问排队等待服务。
　　分布式限流
　　单机限流，可以用到 AtomicInteger、RateLimiter、Semaphore 这些。但是在分布式中，就不能使用了。常用分布式限流用 Nginx 限流，但是它属于网关层面，不能解决所有问题，例如内部服务，短信接口，你无法保证消费方是否会做好限流控制，所以自己在应用层实现限流还是很有必要的。
　　本文不涉及 nginx + lua，简单介绍 redis + lua分布式限流的实现。如果是需要在接入层限流的话，应该直接采用nginx自带的连接数限流模块和请求限流模块。

Redis + Lua 限流示例
　　本次项目使用SpringBoot 2.0.4，使用到 Redis 集群，Lua 限流脚本

引入依赖
　　

　　
　　org.springframework.boot
　　spring-boot-starter-web
　　
　　
　　org.springframework.boot
　　spring-boot-starter-data-redis
　　
　　
　　org.springframework.boot
　　spring-boot-starter-aop
　　
　　
　　org.apache.commons
　　commons-lang3
　　
　　
　　org.springframework.boot
　　spring-boot-starter-test
　　
　　

　　

Redis 配置
　　application.properties
　　

spring.application.name=spring-boot-limit　　

　　
# Redis数据库索引
　　
spring.redis.database=0
　　
# Redis服务器地址
　　
spring.redis.host=10.4.89.161
　　
# Redis服务器连接端口
　　
spring.redis.port=6379
　　
# Redis服务器连接密码（默认为空）
　　
spring.redis.password=
　　
# 连接池最大连接数（使用负值表示没有限制）
　　
spring.redis.jedis.pool.max-active=8
　　
# 连接池最大阻塞等待时间（使用负值表示没有限制）
　　
spring.redis.jedis.pool.max-wait=-1
　　
# 连接池中的最大空闲连接
　　
spring.redis.jedis.pool.max-idle=8
　　
# 连接池中的最小空闲连接
　　
spring.redis.jedis.pool.min-idle=0
　　
# 连接超时时间（毫秒）
　　
spring.redis.timeout=10000
　　

Lua 脚本
　　参考： 聊聊高并发系统之限流特技
　　http://jinnianshilongnian.iteye.com/blog/2305117
　　

local key = "rate.limit:" .. KEYS[1] --限流KEY　　
local limit = tonumber(ARGV[1])        --限流大小
　　
local current = tonumber(redis.call('get', key) or "0")
　　
if current + 1 > limit then --如果超出限流大小
　　return 0
　　
else  --请求数+1，并设置2秒过期
　　redis.call("INCRBY", key,"1")
　　redis.call("expire", key,"2")
　　return current + 1
　　
end
　　

　　1、我们通过KEYS[1] 获取传入的key参数
　　2、通过ARGV[1]获取传入的limit参数
　　3、redis.call方法，从缓存中get和key相关的值，如果为nil那么就返回0
　　4、接着判断缓存中记录的数值是否会大于限制大小，如果超出表示该被限流，返回0
　　5、如果未超过，那么该key的缓存值+1，并设置过期时间为1秒钟以后，并返回缓存值+1

限流注解
　　注解的目的，是在需要限流的方法上使用
　　

package com.souyunku.example.annotation;　　
/**
　　* 描述: 限流注解
　　*
　　* @author yanpenglei
　　* @create 2018-08-16 15:24
　　**/
　　
@Target({ElementType.TYPE, ElementType.METHOD})
　　
@Retention(RetentionPolicy.RUNTIME)
　　
public @interface RateLimit {
　　

　　/**
　　* 限流唯一标示
　　*
　　* @return
　　*/
　　String key() default "";
　　

　　/**
　　* 限流时间
　　*
　　* @return
　　*/
　　int time();
　　

　　/**
　　* 限流次数
　　*
　　* @return
　　*/
　　int count();
　　
}
　　

公共配置
　　

package com.souyunku.example.config;　　

　　
@Component

　　
public>　　

　　/**
　　* 读取限流脚本
　　*
　　* @return
　　*/
　　@Bean
　　public DefaultRedisScript redisluaScript() {
　　DefaultRedisScript redisScript = new DefaultRedisScript();

　　redisScript.setScriptSource(new ResourceScriptSource(new>　　redisScript.setResultType(Number.class);
　　return redisScript;
　　}
　　

　　/**
　　* RedisTemplate
　　*
　　* @return
　　*/
　　@Bean
　　public RedisTemplate limitRedisTemplate(LettuceConnectionFactory redisConnectionFactory) {
　　RedisTemplate template = new RedisTemplate();
　　template.setKeySerializer(new StringRedisSerializer());
　　template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
　　template.setConnectionFactory(redisConnectionFactory);
　　return template;
　　}
　　

　　
}
　　

拦截器
　　通过拦截器 拦截@RateLimit注解的方法，使用Redsi execute 方法执行我们的限流脚本，判断是否超过限流次数
　　以下下是核心代码
　　

package com.souyunku.example.config;　　

　　
/**
　　* 描述:拦截器
　　*
　　* @author yanpenglei
　　* @create 2018-08-16 15:33
　　**/
　　
@Aspect
　　
@Configuration

　　
public>　　

　　private static final Logger logger = LoggerFactory.getLogger(LimitAspect.class);
　　

　　@Autowired
　　private RedisTemplate limitRedisTemplate;
　　

　　@Autowired
　　private DefaultRedisScript redisluaScript;
　　

　　@Around("execution(* com.souyunku.example.controller ..*(..) )")
　　public Object interceptor(ProceedingJoinPoint joinPoint) throws Throwable {
　　

　　MethodSignature signature = (MethodSignature) joinPoint.getSignature();
　　Method method = signature.getMethod();
　　Class targetClass = method.getDeclaringClass();
　　RateLimit rateLimit = method.getAnnotation(RateLimit.class);
　　

　　if (rateLimit != null) {
　　HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
　　String ipAddress = getIpAddr(request);
　　

　　StringBuffer stringBuffer = new StringBuffer();
　　stringBuffer.append(ipAddress).append("-")
　　.append(targetClass.getName()).append("- ")
　　.append(method.getName()).append("-")
　　.append(rateLimit.key());
　　

　　List keys = Collections.singletonList(stringBuffer.toString());
　　

　　Number number = limitRedisTemplate.execute(redisluaScript, keys, rateLimit.count(), rateLimit.time());
　　

　　if (number != null && number.intValue() != 0 && number.intValue()  15) { // "***.***.***.***".length()
　　// = 15
　　if (ipAddress.indexOf(",") > 0) {
　　ipAddress = ipAddress.substring(0, ipAddress.indexOf(","));
　　}
　　}
　　} catch (Exception e) {
　　ipAddress = "";
　　}
　　return ipAddress;
　　}
　　
}
　　

　　

控制层
　　添加 @RateLimit() 注解，会在 Redsi 中生成 10 秒中，可以访问5次 的key
　　RedisAtomicLong 是为测试例子例，记录累计访问次数，跟限流没有关系。
　　

package com.souyunku.example.controller;　　

　　
/**
　　* 描述: 测试页
　　*
　　* @author yanpenglei
　　* @create 2018-08-16 15:42
　　**/
　　
@RestController

　　
public>　　

　　@Autowired
　　private RedisTemplate redisTemplate;
　　

　　// 10 秒中，可以访问10次
　　@RateLimit(key = "test", time = 10, count = 10)
　　@GetMapping("/test")
　　public String luaLimiter() {
　　RedisAtomicInteger entityIdCounter = new RedisAtomicInteger("entityIdCounter", redisTemplate.getConnectionFactory());
　　

　　String date = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
　　

　　return date + " 累计访问次数：" + entityIdCounter.getAndIncrement();
　　}
　　
}
　　

启动服务
　　

package com.souyunku.example;　　

　　
@SpringBootApplication

　　
public>　　

　　public static void main(String[] args) {
　　SpringApplication.run(SpringBootLimitApplication.class, args);
　　}
　　
}
　　

　　启动项目页面访问：http://127.0.0.1:8080/test
　　10 秒中，可以访问10次,超过十次，页面就报错，等够10秒，重新计算。
　　后台日志
　　

2018-08-16 18:41:08.205  INFO 18076 --- [nio-8080-exec-1] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：1 次　　
2018-08-16 18:41:08.426  INFO 18076 --- [nio-8080-exec-3] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：2 次
　　
2018-08-16 18:41:08.611  INFO 18076 --- [nio-8080-exec-5] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：3 次
　　
2018-08-16 18:41:08.819  INFO 18076 --- [nio-8080-exec-7] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：4 次
　　
2018-08-16 18:41:09.021  INFO 18076 --- [nio-8080-exec-9] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：5 次
　　
2018-08-16 18:41:09.203  INFO 18076 --- [nio-8080-exec-1] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：6 次
　　
2018-08-16 18:41:09.406  INFO 18076 --- [nio-8080-exec-3] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：7 次
　　
2018-08-16 18:41:09.629  INFO 18076 --- [nio-8080-exec-5] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：8 次
　　
2018-08-16 18:41:09.874  INFO 18076 --- [nio-8080-exec-7] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：9 次
　　
2018-08-16 18:41:10.178  INFO 18076 --- [nio-8080-exec-9] com.souyunku.example.config.LimitAspect  : 限流时间段内访问第：10 次
　　
2018-08-16 18:41:10.702 ERROR 18076 --- [nio-8080-exec-1] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is java.lang.RuntimeException: 已经到设置限流次数] with root cause
　　

　　
java.lang.RuntimeException: 已经到设置限流次数
　　at com.souyunku.example.config.LimitAspect.interceptor(LimitAspect.java:73) ~[classes/:na]
　　at sun.reflect.GeneratedMethodAccessor35.invoke(Unknown Source) ~[na:na]
　　at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_112]
　　at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_112]
　　

推荐阅读

• Dubbo 整合 Pinpoint 做分布式服务请求跟踪
  
• 接口限流：漏桶算法&令牌桶算法
  
• Java并发：分布式应用限流实践
  
• Java并发：Semaphore信号量源码分析
  
• Java并发：深入浅出AQS之共享锁模式源码分析
  
• Java并发：深入浅出AQS之独占锁模式源码分析
  
• Java并发：了解无锁CAS就从源码分析
  
• Java并发：CAS原理分析
  

Contact

• 作者：鹏磊
  
• 出处：http://www.ymq.io/2018/08/16/redis-lua/
  
• 版权归作者所有，转载请注明出处
  
• Wechat：关注公众号，搜云库，专注于开发技术的研究与知识分享