中小型项目请求限流设计

Posted 2023-04-11 valarchie

何为请求限流？

请求限流是一种控制API或其他Web服务的流量的技术。它的目的是限制客户端对服务器发出的请求的数量或速率，以防止服务器过载或响应时间变慢，从而提高系统的可用性和稳定性。

中小型项目请求限流的需求

1. 按IP、用户、全局限流
2. 基于不同实现的限流设计（基于Redis或者LRU缓存）
3. 基于注解标注哪些接口限流

完整限流设计实现在开源项目中：https://github.com/valarchie/AgileBoot-Back-End

注解设计

声明一个注解类，主要有以下几个属性

• key(缓存的key)
• time(时间范围)
• maxCount(时间范围内最大的请求次数)
• limitType(按IP/用户/全局进行限流)
• cacheType(基于Redis或者Map来实现限流)

/**
 * 限流注解
 *
 * @author valarchie
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimit 

    /**
     * 限流key
     */
    String key() default "None";

    /**
     * 限流时间,单位秒
     */
    int time() default 60;

    /**
     * 限流次数
     */
    int maxCount() default 100;

    /**
     * 限流条件类型
     */
    LimitType limitType() default LimitType.GLOBAL;

    /**
     * 限流使用的缓存类型
     */
    CacheType cacheType() default CacheType.REDIS;

LimitType枚举，我们可以将不同限制类型的逻辑直接放在枚举类当中。推荐将逻辑直接放置在枚举类中，代码的组织形式会更好。

enum LimitType 
        /**
         * 默认策略全局限流  不区分IP和用户
         */
        GLOBAL
            @Override
            public String generateCombinedKey(RateLimit rateLimiter) 
                return rateLimiter.key() + this.name();
            
        ,

        /**
         * 根据请求者IP进行限流
         */
        IP 
            @Override
            public String generateCombinedKey(RateLimit rateLimiter) 
                String clientIP = ServletUtil.getClientIP(ServletHolderUtil.getRequest());
                return rateLimiter.key() + clientIP;
            
        ,

        /**
         * 按用户限流
         */
        USER 
            @Override
            public String generateCombinedKey(RateLimit rateLimiter) 
                LoginUser loginUser = AuthenticationUtils.getLoginUser();
                if (loginUser == null) 
                    throw new ApiException(ErrorCode.Client.COMMON_NO_AUTHORIZATION);
                
                return rateLimiter.key() + loginUser.getUsername();
            
        ;

        public abstract String generateCombinedKey(RateLimit rateLimiter);

    
    

CacheType, 主要分为Redis和Map, 后续有新的类型可以新增。

  enum CacheType 

      /**
       * 使用redis做缓存
       */
      REDIS,

      /**
       * 使用map做缓存
       */
      Map

  

RateLimitChecker设计

声明一个抽象类，然后将具体实现放在实现类中，便于扩展

/**
 * @author valarchie
 */
public abstract class AbstractRateLimitChecker 

    /**
     * 检查是否超出限流
     * @param rateLimiter
     */
    public abstract void check(RateLimit rateLimiter);

Redis限流实现

/**
 * @author valarchie
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class RedisRateLimitChecker extends AbstractRateLimitChecker

    @NonNull
    private RedisTemplate<Object, Object> redisTemplate;

    private final RedisScript<Long> limitScript = new DefaultRedisScript<>(limitScriptText(), Long.class);

    @Override
    public void check(RateLimit rateLimiter) 
        int maxCount = rateLimiter.maxCount();
        String combineKey = rateLimiter.limitType().generateCombinedKey(rateLimiter);

        Long currentCount;
        try 
            currentCount = redisTemplate.execute(limitScript, ListUtil.of(combineKey), maxCount, rateLimiter.time());
            log.info("限制请求:, 当前请求次数:, 缓存key:", combineKey, currentCount, rateLimiter.key());
         catch (Exception e) 
            throw new RuntimeException("redis限流器异常，请确保redis启动正常");
        

        if (currentCount == null) 
            throw new RuntimeException("redis限流器异常，请稍后再试");
        

        if (currentCount.intValue() > maxCount) 
            throw new ApiException(ErrorCode.Client.COMMON_REQUEST_TOO_OFTEN);
        

    

    /**
     * 限流脚本
     */
    private static String limitScriptText() 
        return "local key = KEYS[1]\\n" +
            "local count = tonumber(ARGV[1])\\n" +
            "local time = tonumber(ARGV[2])\\n" +
            "local current = redis.call(\'get\', key);\\n" +
            "if current and tonumber(current) > count then\\n" +
            "    return tonumber(current);\\n" +
            "end\\n" +
            "current = redis.call(\'incr\', key)\\n" +
            "if tonumber(current) == 1 then\\n" +
            "    redis.call(\'expire\', key, time)\\n" +
            "end\\n" +
            "return tonumber(current);";
    

Map + Guava RateLimiter实现

/**
 * @author valarchie
 */
@SuppressWarnings("UnstableApiUsage")
@Component
@RequiredArgsConstructor
@Slf4j
public class MapRateLimitChecker extends AbstractRateLimitChecker

    /**
     * 最大仅支持4096个key   超出这个key  限流将可能失效
     */
    private final LRUCache<String, RateLimiter> cache = new LRUCache<>(4096);


    @Override
    public void check(RateLimit rateLimit) 
        String combinedKey = rateLimit.limitType().generateCombinedKey(rateLimit);

        RateLimiter rateLimiter = cache.get(combinedKey,
            () -> RateLimiter.create((double) rateLimit.maxCount() / rateLimit.time())
        );

        if (!rateLimiter.tryAcquire()) 
            throw new ApiException(ErrorCode.Client.COMMON_REQUEST_TOO_OFTEN);
        

        log.info("限制请求key:, combined key:", rateLimit.key(), combinedKey);
    

限流切面

我们需要在切面中，读取限流注解标注的信息，然后选择不同的限流实现来进行限流。

/**
 * 限流切面处理
 *
 * @author valarchie
 */
@Aspect
@Component
@Slf4j
@ConditionalOnExpression("\'$agileboot.embedded.redis\' != \'true\'")
@RequiredArgsConstructor
public class RateLimiterAspect 

    @NonNull
    private RedisRateLimitChecker redisRateLimitChecker;

    @NonNull
    private MapRateLimitChecker mapRateLimitChecker;


    @Before("@annotation(rateLimiter)")
    public void doBefore(JoinPoint point, RateLimit rateLimiter) 
        MethodSignature signature = (MethodSignature) point.getSignature();
        Method method = signature.getMethod();
        log.info("当前限流方法:" + method.toGenericString());

        switch (rateLimiter.cacheType()) 
            case REDIS:
                redisRateLimitChecker.check(rateLimiter);
                break;
            case Map:
                mapRateLimitChecker.check(rateLimiter);
                return;
            default:
                redisRateLimitChecker.check(rateLimiter);
        

    

注解使用

以下是我们标注的注解例子。

time=10,maxCount=10表明10秒内最多10次请求。

cacheType=Redis表明使用Redis来实现。

limitType=IP表明基于IP来限流。

/**
 * 生成验证码
 */
@Operation(summary = "验证码")
@RateLimit(key = RateLimitKey.LOGIN_CAPTCHA_KEY, time = 10, maxCount = 10, cacheType = CacheType.REDIS,
    limitType = LimitType.IP)
@GetMapping("/captchaImage")
public ResponseDTO<CaptchaDTO> getCaptchaImg() 
    CaptchaDTO captchaImg = loginService.generateCaptchaImg();
    return ResponseDTO.ok(captchaImg);

这是笔者关于中小型项目关于请求限流的实现，如有不足欢迎大家评论指正。

全栈技术交流群：1398880

经典面试题——让你设计一个限流的系统怎么做？

参考技术A 保障服务稳定的三大利器：熔断降级、服务限流和故障模拟。限流系统是当前很多系统都需要考虑的场景。首先在Nginx层面是可以做限流的，除此之外，在微服务层面还是有很多空间可以施展的。

限流的话，主要思路分为 请求数量限制 和 消费能力限制 两类。前者主要是限制一段时间内的总并发数、后者主要是限制消费者的消费能力。

限流算法

限流算法来说，主要包含令牌桶算法、漏桶算法和计数器等。 对于简单的计数器算法 ，通过AtomicLong#incrementAndGet()来进行粗暴的控制，因为容易导致“突刺现象”（比如单位时间1s内的前10ms，已经通过了100个请求，那后面的990ms，只能拒绝）， 所以这里不做推荐 。

令牌桶算法

令牌工厂：匀速生成令牌

令牌桶：拥有固定的令牌数

应用者：一次可以申请N个令牌，没有令牌不能进行后续处理。

如果使用Redis来实现的话会比较简单，大概思路如下：

1/ 获取令牌：依靠List的leftPop来获取令牌

Objectresult = redisTemplate.opsForList().leftPop("limit_list");

2/ 向令牌桶添加令牌

redisTemplate.opsForList().rightPush("limit_list",UUID.randomUUID().toString());

漏桶算法

漏桶：容量固定

流速：任意

流出水滴：固定速率

滑动窗口

用Redis的list数据结构可以轻而易举的实现该功能：保证每N秒内至多M个请求，缺点就是zset的数据结构会越来越大。实现方式相对也是比较简单的。

1/ 请求进来：UUID生成唯一的value；score用当前的时间戳

redisTemplate.opsForZSet().add("limit",UUID.randomUUID().toString(),currentTime);

2/ 限流：zset的range方法可以统计两个时间戳内的请求，达到限流效果

Integercount = redisTemplate.opsForZSet().rangeByScore("limit", currentTime -