阅读完需:约 9 分钟
Redis 除了做缓存,还能干很多很多事情:分布式锁、限流、处理请求接口幂等性。
在项目中因为某个接口疯狂被请求导致线程阻塞,所以要对单个接口加上阻塞。
准备工作
添加POM依赖
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.4.RELEASE</version>
</parent>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
然后连接redis即可
限流注解
接下来我们创建一个限流注解,我们将限流分为两种情况:
- 针对当前接口的全局性限流,例如该接口可以在 1 分钟内访问 100 次。
- 针对某一个 IP 地址的限流,例如某个 IP 地址可以在 1 分钟内访问 100 次。
针对这两种情况,我们创建一个枚举类:
public enum LimitType {
/**
* 默认策略全局限流
*/
DEFAULT,
/**
* 根据请求者IP进行限流
*/
IP
}
接下来我们来创建限流注解:
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
/**
* 限流key
*/
String key() default "rate_limit:";
/**
* 限流时间,单位秒
*/
int time() default 60;
/**
* 限流次数
*/
int count() default 100;
/**
* 限流类型
*/
LimitType limitType() default LimitType.DEFAULT;
}
第一个参数限流的 key,这个仅仅是一个前缀,将来完整的 key 是这个前缀再加上接口方法的完整路径,共同组成限流 key,这个 key 将被存入到 Redis 中
将来哪个接口需要限流,就在哪个接口上添加 @RateLimiter
注解
定制 RedisTemplate
在 Spring Boot 中,我们更习惯使用 Spring Data Redis 来操作 Redis,不过默认的 RedisTemplate
有一个小坑,就是序列化用的是 JdkSerializationRedisSerializer
,不知道小伙伴们有没有注意过,直接用这个序列化工具将来存到 Redis 上的 key 和 value 都会莫名其妙多一些前缀,这就导致你用命令读取的时候可能会出错。
例如存储的时候,key 是 name,value 是 javaboy,但是当你在命令行操作的时候,get name
却获取不到你想要的数据,原因就是存到 redis 之后 name 前面多了一些字符,此时只能继续使用 RedisTemplate 将之读取出来。
我们用 Redis 做限流会用到 Lua 脚本,使用 Lua 脚本的时候,就会出现上面说的这种情况,所以我们需要修改 RedisTemplate 的序列化方案
为什么不用 StringRedisTemplate 呢?StringRedisTemplate 确实不存在上面所说的问题,但是它能够存储的数据类型不够丰富,所以这里不考虑
修改 RedisTemplate 序列化方案,代码如下:
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(connectionFactory);
// 使用Jackson2JsonRedisSerialize 替换默认序列化(默认采用的是JDK序列化)
Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
redisTemplate.setKeySerializer(jackson2JsonRedisSerializer);
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
redisTemplate.setHashKeySerializer(jackson2JsonRedisSerializer);
redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
return redisTemplate;
}
}
key 和 value 我们都使用 Spring Boot 中默认的 jackson 序列化方式来解决。
开发 Lua 脚本
Redis 中的一些原子操作我们可以借助 Lua 脚本来实现,想要调用 Lua 脚本,我们有两种不同的思路:
- 在 Redis 服务端定义好 Lua 脚本,然后计算出来一个散列值,在 Java 代码中,通过这个散列值锁定要执行哪个 Lua 脚本。
- 直接在 Java 代码中将 Lua 脚本定义好,然后发送到 Redis 服务端去执行。
Spring Data Redis 中也提供了操作 Lua 脚本的接口,还是比较方便的,所以我们这里就采用第二种方案。
我们在 resources 目录下新建 lua 文件夹专门用来存放 lua 脚本,脚本内容如下:
local key = KEYS[1]
local count = tonumber(ARGV[1])
local time = tonumber(ARGV[2])
local current = redis.call('get', key)
if current and tonumber(current) > count then
return tonumber(current)
end
current = redis.call('incr', key)
if tonumber(current) == 1 then
redis.call('expire', key, time)
end
return tonumber(current)
KEYS 和 ARGV 都是一会调用时候传进来的参数,tonumber 就是把字符串转为数字,redis.call 就是执行具体的 redis 指令,具体流程是这样:
- 首先获取到传进来的 key 以及 限流的 count 和时间 time。
- 通过 get 获取到这个 key 对应的值,这个值就是当前时间窗内这个接口可以访问多少次。
- 如果是第一次访问,此时拿到的结果为 nil,否则拿到的结果应该是一个数字,所以接下来就判断,如果拿到的结果是一个数字,并且这个数字还大于 count,那就说明已经超过流量限制了,那么直接返回查询的结果即可。
- 如果拿到的结果为 nil,说明是第一次访问,此时就给当前 key 自增 1,然后设置一个过期时间。
- 最后把自增 1 后的值返回就可以了。
接下来我们在一个 Bean 中来加载这段 Lua 脚本,如下:
@Bean
public DefaultRedisScript<Long> limitScript() {
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/limit.lua")));
redisScript.setResultType(Long.class);
return redisScript;
}
注解解析
接下来我们就需要自定义切面,来解析这个注解了,我们来看看切面的定义:
@Aspect
@Component
public class RateLimiterAspect {
private static final Logger log = LoggerFactory.getLogger(RateLimiterAspect.class);
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
@Autowired
private RedisScript<Long> limitScript;
@Before("@annotation(rateLimiter)")
public void doBefore(JoinPoint point, RateLimiter rateLimiter) throws Throwable {
String key = rateLimiter.key();
int time = rateLimiter.time();
int count = rateLimiter.count();
String combineKey = getCombineKey(rateLimiter, point);
List<Object> keys = Collections.singletonList(combineKey);
try {
Long number = redisTemplate.execute(limitScript, keys, count, time);
if (number==null || number.intValue() > count) {
throw new LimiterException("访问过于频繁,请稍候再试");
}
log.info("限制请求'{}',当前请求'{}',缓存key'{}'", count, number.intValue(), key);
} catch (RuntimeException e) {
throw e;
}
}
public String getCombineKey(RateLimiter rateLimiter, JoinPoint point) {
StringBuffer stringBuffer = new StringBuffer(rateLimiter.key());
if (rateLimiter.limitType() == LimitType.IP) {
stringBuffer.append(IpUtil.getIp()).append("-");
}
MethodSignature signature = (MethodSignature) point.getSignature();
Method method = signature.getMethod();
Class<?> targetClass = method.getDeclaringClass();
stringBuffer.append(targetClass.getName()).append("-").append(method.getName());
return stringBuffer.toString();
}
}
这个切面就是拦截所有加了 @RateLimiter
注解的方法,在前置通知中对注解进行处理。
- 首先获取到注解中的 key、time 以及 count 三个参数。
- 获取一个组合的 key,所谓的组合的 key,就是在注解的 key 属性基础上,再加上方法的完整路径,如果是 IP 模式的话,就再加上 IP 地址。以 IP 模式为例,最终生成的 key 类似这样:
rate_limit:127.0.0.1-org.javaboy.ratelimiter.controller.HelloController-hello
(如果不是 IP 模式,那么生成的 key 中就不包含 IP 地址)。 - 将生成的 key 放到集合中。
- 通过 redisTemplate.execute 方法取执行一个 Lua 脚本,第一个参数是脚本所封装的对象,第二个参数是 key,对应了脚本中的 KEYS,后面是可变长度的参数,对应了脚本中的 ARGV。
- 将 Lua 脚本执行的结果与 count 进行比较,如果大于 count,就说明过载了,抛异常就行了。
解析中关于IP的获取工具类
IpUtil
@Slf4j
public class IpUtil {
// 多次反向代理后会有多个ip值 的分割符
private static final String IP_UTILS_FLAG = ",";
// 未知IP
private static final String UNKNOWN = "unknown";
// 本地 IP
private static final String LOCALHOST_IP = "0:0:0:0:0:0:0:1";
private static final String LOCALHOST_IP1 = "127.0.0.1";
public static String getIp(){
// 根据 HttpHeaders 获取 请求 IP地址
HttpServletRequest request = ((ServletRequestAttributes) Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest();
String ip = request.getHeader("X-Forwarded-For");
if (StringUtils.isEmpty(ip) || UNKNOWN.equalsIgnoreCase(ip)) {
ip = request.getHeader("x-forwarded-for");
if (ip != null && ip.length() != 0 && !UNKNOWN.equalsIgnoreCase(ip)) {
// 多次反向代理后会有多个ip值,第一个ip才是真实ip
if (ip.contains(IP_UTILS_FLAG)) {
ip = ip.split(IP_UTILS_FLAG)[0];
}
}
}
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.getHeader("HTTP_CLIENT_IP");
}
if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
ip = request.getHeader("HTTP_X_FORWARDED_FOR");
}
if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
ip = request.getHeader("X-Real-IP");
}
//兼容k8s集群获取ip
if (StringUtils.isEmpty(ip) || UNKNOWN.equalsIgnoreCase(ip)) {
ip = Objects.requireNonNull(request.getRemoteAddr());
if (LOCALHOST_IP1.equalsIgnoreCase(ip) || LOCALHOST_IP.equalsIgnoreCase(ip)) {
//根据网卡取本机配置的IP
InetAddress iNet = null;
try {
iNet = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
log.error("getClientIp error: ", e);
}
assert iNet != null;
ip = iNet.getHostAddress();
}
}
return ip;
}
}
异常类创建
public class LimiterException extends RuntimeException{
public LimiterException(String message) {
super(message);
}
}
@ExceptionHandler(LimiterException.class)
public Map<String,Object> serviceException(LimiterException e) {
HashMap<String, Object> map = new HashMap<>();
map.put("code", 11111);
map.put("message", e.getMessage());
map.put("success",false);
return map;
}
接口测试
@RestController
public class HelloController {
@GetMapping("/hello")
@RateLimiter(time = 5,count = 3,limitType = LimitType.IP)
public String hello() {
return "hello>>>"+new Date();
}
}
结果
{
"code": 11111,
"success": false,
"message": "访问过于频繁,请稍候再试"
}