分布式环境下，基于Redis实现Restful API接口的限流

╰半橙微兮° 2021-09-24 12:20 296阅读 0赞

> `创作不易，如果觉得这篇文章对你有帮助，欢迎各位老铁点个赞支持下呗，您的支持是我创作的最大动力！`

### 文章目录 ###

*  1 前言
 *  2 为什么要对Api接口限流
 *  3 限流方案的选择
 *  4 基于Redis实现限流
 *   *  4.1 实现的思路
     *  4.2 实现限流
     *   *  4.2.1 定义限流注解
         *  4.2.2 定义切面类，拦截需要限流的方法
         *  4.2.3 业务方法添加限流注解
     *  4.3 对以上示例简要分析

# 1 前言 #

随着时代的发展，互联网也发生了巨大的变化。其中重要的一个变化时，`为了应对高流量，服务的架构从集中式架构，演变成了分布式架构`。

**什么是分布式架构？**  
简单来说，就是之前的`一个单体应用（后台管理系统）`，通过拆分，`拆分成用户中心、产品中心、客户中心等多个小应用服务`，这种`把一个大的单体应用项目，拆分成多个小应用项目的方式，就是分布式系统应用架构`。

本文将以一种最高效的方式，实现分布式环境下，接口方法的限流。

# 2 为什么要对Api接口限流 #

为了满足各种应用场景，有时候不得不对接口Api进行限流。比如说：短信服务，供应商可能会要求每秒访问不超过400条，如果超过了这个访问量，请求就会被供应商拒绝，从而导致漏发短信。

还有的接口，第三方Api会做限制，他们为了限制访问，设定一分钟只能请求接口20次，超过了就会超时或者响应异常。

总而言之，限流，在好多场景用的还是挺多的。

# 3 限流方案的选择 #

限流，方案有很多，具体的典型的限流方案介绍，请参考我的另一篇博文：[分布式环境下限流方案的思考][Link 1]

最终选择的方案是基于Redis实现的限流，该方案是目前最流行，也是最高效的一种方式。

# 4 基于Redis实现限流 #

## 4.1 实现的思路 ##

**思路：** 借助于`Redis`的`INCR`操作来实现Limit限流

*  将INCR key中储存的数字值增一，如果key不存在，那么key的值会先被初始化为 0 ，然后再执行INCR操作。
    
    如果值包含错误的类型，或字符串类型的值不能表示为数字，那么返回一个错误，本操作的值限制在 64 位(bit)有符号数字表示之内。
 *  当API被调用时，在调用API前进行INCR key，key可以是ip地址相关，用户相关，`业务参数相关`，或是`全局的一个key`。如果返回值为1，则表示刚开始调用，赋予key过期时间，然后判断返回值是否大于设定的Limit限流数量，如果大于抛异常或者阻塞重试。

## 4.2 实现限流 ##

### 4.2.1 定义限流注解 ###

**代码示例如下：**

/** * * 自定义，限流注解（默认一分钟，限流500次） * * * @author smilehappiness * @Date 2020/7/5 20:05 */
 @Order(Ordered.HIGHEST_PRECEDENCE)
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.METHOD)
 @Documented
 public @interface ApiLimit { 
 
 /** * 限制某时间段内可以访问的次数，默认设置500 */
 int limitCounts() default 500;
 
 /** * @return * @Description: 限制访问的某一个时间段，单位为秒，默认值1分钟 */
 int timeSecond() default 60;
 }

### 4.2.2 定义切面类，拦截需要限流的方法 ###

**代码示例如下：**

package cn.smilehappiness.aspect;
 
 import cn.smilehappiness.annotation.ApiLimit;
 import cn.smilehappiness.model.SmsMessage;
 import com.alibaba.fastjson.JSON;
 import org.aspectj.lang.JoinPoint;
 import org.aspectj.lang.ProceedingJoinPoint;
 import org.aspectj.lang.Signature;
 import org.aspectj.lang.annotation.Around;
 import org.aspectj.lang.annotation.Aspect;
 import org.aspectj.lang.annotation.Pointcut;
 import org.aspectj.lang.reflect.MethodSignature;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.ValueOperations;
 import org.springframework.stereotype.Component;
 import org.springframework.web.context.request.RequestContextHolder;
 import org.springframework.web.context.request.ServletRequestAttributes;
 
 import javax.servlet.http.HttpServletRequest;
 import java.lang.reflect.Method;
 import java.util.concurrent.TimeUnit;
 
 
 /** * Api Limit切面类 * * @author smilehappiness * @Date 2020/7/5 19:55 */
 @Aspect
 @Component
 public class ApiLimitAspect { 
 
 private Logger logger = LoggerFactory.getLogger(getClass());
 
 /** * 通过构造注入的方式，注入redisTemplate */
 private final RedisTemplate<String, Object> redisTemplate;
 
 public ApiLimitAspect(RedisTemplate<String, Object> redisTemplate) { 
 this.redisTemplate = redisTemplate;
 }
 
 /** * * 定义切面表达式（类的维度拦截，可以拦截更多的方法） * * * @param * @return void * @Date 2020/7/5 20:07 */
 @Pointcut("execution(* cn.smilehappiness..service..*Impl.*(..))")
 private void myPointCut() { 
 }
 
 /** * * 实现对添加ApiLimit注解的方法进行拦截，Limit处理（方式一：切面类拦截） * * * @param joinPoint * @return void * @Date 2020/7/5 20:10 */
 //@Around("myPointCut()")
 public void requestLimit(ProceedingJoinPoint joinPoint) throws Throwable { 
 ApiLimit apiLimit = this.getAnnotation(joinPoint);
 //对业务方法进行全局限流
 if (apiLimit != null) { 
 dealLimit(apiLimit, joinPoint, false);
 }
 }
 
 /** * * 针对业务方法进行Limit限流处理，如果第一次请求被限制了，等待10秒后重试，如果再次失败，则抛出异常 * * * @param apiLimit * @param joinPoint * @param flag 判断是否进行过一次重试了，如果重试过一次还是被限制，就抛异常 * @return void * @Date 2020/7/5 20:30 */
 private void dealLimit(ApiLimit apiLimit, ProceedingJoinPoint joinPoint, Boolean flag) throws Throwable { 
 String msgKey = checkParam(apiLimit, joinPoint);
 //业务方法中，参数唯一key
 String cacheKey = "smsService:sendLimit:" + msgKey;
 
 ValueOperations<String, Object> valueOperations = redisTemplate.opsForValue();
 long methodCounts = valueOperations.increment(cacheKey, 1);
 // 如果该key不存在，则从0开始计算，并且当count为1的时候，设置过期时间
 if (methodCounts == 1) { 
 redisTemplate.expire(cacheKey, apiLimit.timeSecond(), TimeUnit.SECONDS);
 }
 
 // 如果redis中的count大于限制的次数，则等待10秒重试
 if (methodCounts > apiLimit.limitCounts()) { 
 if (!flag) { 
 //等待10秒后，第一次重试
 Thread.sleep(10 * 1000);
 logger.warn("等待10秒后，第一次重试...");
 
 // 递归，再次请求业务方法
 dealLimit(apiLimit, joinPoint, true);
 } else { 
 //如果第一次请求被限制了，等待10秒后重试，如果再次失败，则抛出异常，当然，如果不需要重试，直接抛异常或者逻辑处理即可
 throw new RuntimeException("短信发送三方Api接口超限，请30秒后再试！");
 }
 } else { 
 //正常执行业务方法
 joinPoint.proceed();
 }
 
 }
 
 /** * * 业务限流方法，业务参数非空检验 * * * @param apiLimit * @param joinPoint * @return java.lang.String * @Date 2020/7/5 20:40 */
 private String checkParam(ApiLimit apiLimit, ProceedingJoinPoint joinPoint) { 
 if (apiLimit == null) { 
 throw new RuntimeException("限流方法dealLimit处理异常!");
 }
 
 //获取方法的参数，通过参数设置缓存的唯一key
 Object[] args = joinPoint.getArgs();
 if (args == null) { 
 throw new RuntimeException("业务方法参数不允许为空！");
 }
 
 SmsMessage smsMessage = null;
 if (args[0] instanceof SmsMessage) { 
 smsMessage = (SmsMessage) args[0];
 }
 if (smsMessage == null) { 
 throw new RuntimeException("HttpServletRequest请求，参数异常!");
 }
 
 // 获取HttpRequest
 ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
 if (attributes == null) { 
 throw new RuntimeException("HttpServletRequest请求，attributes参数异常!");
 }
 
 HttpServletRequest request = attributes.getRequest();
 // 判断request不能为空
 if (request == null) { 
 throw new RuntimeException("HttpServletRequest请求，参数异常!");
 }
 
 String ip = request.getRemoteAddr();
 String uri = request.getRequestURI();
 logger.debug("请求的参数ip：【{}】, uri：【{}】,请求参数：【{}】", ip, uri, JSON.toJSONString(smsMessage));
 
 return smsMessage.getMsgKey();
 }
 
 /** * * 环绕通知，对使用ApiLimit注解的方法进行拦截，限流处理（方式二：直接对注解进行拦截） * * * @param joinPoint * @return void * @Date 2020/7/5 21:03 */
 @Around("@annotation(cn.smilehappiness.annotation.ApiLimit)")
 public void requestLimitByAnnotation(ProceedingJoinPoint joinPoint) throws Throwable { 
 MethodSignature signature = (MethodSignature) joinPoint.getSignature();
 Method method = signature.getMethod();
 if (method == null) { 
 return;
 }
 
 if (method.isAnnotationPresent(ApiLimit.class)) { 
 dealLimit(method.getAnnotation(ApiLimit.class), joinPoint, false);
 }
 }
 
 /** * * 获取注解 * * * @param joinPoint * @return cn.smilehappiness.annotation.ApiLimit * @Date 2020/7/5 21:45 */
 private ApiLimit getAnnotation(JoinPoint joinPoint) { 
 Signature signature = joinPoint.getSignature();
 MethodSignature methodSignature = (MethodSignature) signature;
 Method method = methodSignature.getMethod();
 if (method == null) { 
 return null;
 }
 
 return method.getAnnotation(ApiLimit.class);
 }
 }

### 4.2.3 业务方法添加限流注解 ###

/** * * 根据消息模板以及内容，发送短信 * 默认一分钟，限流500次，可以根据实际情况进行限流 * * * @param smsMessage * @return void * @Date 2020/7/6 21:35 */
 @ApiLimit(limitCounts = 10, timeSecond = 120)
 @Override
 public void sendSmsMessage(SmsMessage smsMessage) { 
 // 注意：一般三方可能会限制，400/s，即每秒最多发送400条，超过这个限制的短信发送请求会被拒绝，所以需要限流，在高流量下，需要在业务端限制，每秒访问不要超过400次
 // 这里只是模拟这种限流的场景，具体的限流大小，根据实际场景去设置
 
 // TODO 调用第三方发送短信
 System.out.println("调用三方短信服务，发送短信成功！");
 }

## 4.3 对以上示例简要分析 ##

主要就就是对需要限流的业务方法，添加了`@ApiLimit(limitCounts = 10, timeSecond = 120)`注解，表示2分钟只允许Api方法被调用10次，否则就会限流。第一次限流进行重试，如果10秒后还不能调用，则抛出异常，待业务端处理。

以上示例中，使用了两种方式进行拦截，一种是`定义切入点`的方式，一种是直接拦截使用`ApiLimit`注解的方法，这两种方式都可以。

完整代码示例，我已分享到GitHub上，需要的童鞋们可以下载：[distributed-limit-api][]

`好啦，本篇到这里就介绍完毕了，有问题的可以评论交流哈，希望对老铁们有所帮助！`

> 写博客是为了记住自己容易忘记的东西，另外也是对自己工作的总结，希望尽自己的努力，做到更好，大家一起努力进步！  
>   
> 如果有什么问题，欢迎大家评论，一起探讨，代码如有问题，欢迎各位大神指正！  
>   
> 给自己的梦想添加一双翅膀，让它可以在天空中自由自在的飞翔！

[Link 1]: https://blog.csdn.net/smilehappiness/article/details/107175286
[distributed-limit-api]: https://github.com/smilehappinessli/distributed-limit-api