SpringBoot - 优雅的实现【流控】

概述限流算法

计数器限流漏桶算法令牌桶算法 V1.0V2.0 自定义注解+AOP实现接口限流

搞依赖搞自定义限流注解搞AOP用上验证 源码

限流 简言之就是当请求达到一定的并发数或速率，就对服务进行等待、排队、降级、拒绝服务等操作。

我们先简单捋一捋限流算法

并发编程-25 高并发处理手段之消息队列思路 + 应用拆分思路 + 应用限流思路

深入理解分布式技术 - 限流

把水比作是请求，漏桶比作是系统处理能力极限，水先进入到漏桶里，漏桶里的水按一定速率流出，当流出的速率小于流入的速率时，由于漏桶容量有限，后续进入的水直接溢出（拒绝请求），以此实现限流

可以简单地理解为医去银行办理业务，只有拿到号以后才可以进行业务办理。

系统会维护一个令牌（token）桶，以一个恒定的速度往桶里放入令牌（token），这时如果有请求进来想要被处理，则需要先从桶里获取一个令牌（token），当桶里没有令牌（token）可取时，则该请求将被拒绝服务。令牌桶算法通过控制桶的容量、发放令牌的速率，来达到对请求的限制。

上 guava

	  
            com.google.guava
            guava
            30.1-jre
        

package com.artisan.controller;

import com.artisan.annos.ArtisanLimit;
import com.google.common.util.concurrent.RateLimiter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;



@Slf4j
@RestController
@RequestMapping("/rateLimit")
public class RateLimitController {

    
    private final RateLimiter limiter = RateLimiter.create(1);

    private DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @SneakyThrows
    @GetMapping("/test")
    public String testLimiter() {
        //500毫秒内，没拿到令牌，就直接进入服务降级
        boolean tryAcquire = limiter.tryAcquire(500, TimeUnit.MILLISECONDS);

        if (!tryAcquire) {
            log.warn("BOOM 服务降级，时间{}", LocalDateTime.now().format(dtf));
            return "系统繁忙，请稍后再试！";
        }

        log.info("获取令牌成功，时间{}", LocalDateTime.now().format(dtf));

        return "业务处理成功";
    }

 

    

我们可以看到RateLimiter的2个核心方法：create()、tryAcquire()

acquire() 获取一个令牌, 改方法会阻塞直到获取到这一个令牌, 返回值为获取到这个令牌花费的时间acquire(int permits) 获取指定数量的令牌, 该方法也会阻塞, 返回值为获取到这 N 个令牌花费的时间tryAcquire() 判断时候能获取到令牌, 如果不能获取立即返回 falsetryAcquire(int permits) 获取指定数量的令牌, 如果不能获取立即返回 falsetryAcquire(long timeout, TimeUnit unit) 判断能否在指定时间内获取到令牌, 如果不能获取立即返回 falsetryAcquire(int permits, long timeout, TimeUnit unit) 同上

测试一下

1.0的功能实现了，但是业务代码和限流代码混在一起，非常的不美观。

 
            org.springframework.boot
            spring-boot-starter-aop
        

package com.artisan.annos;

import java.lang.annotation.*;
import java.util.concurrent.TimeUnit;




@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@documented
public @interface ArtisanLimit {

    
    String key() default "";

    
    double permitsPerSecond();

    
    long timeout();

    
    TimeUnit timeunit() default TimeUnit.MILLISECONDS;

    
    String message() default "系统繁忙,请稍后再试.";
}
    

使用AOP切面拦截限流注解

package com.artisan.aop;

import com.artisan.annos.ArtisanLimit;
import com.artisan.resp.ResponseCode;
import com.artisan.resp.ResponseData;
import com.artisan.utils.WebUtils;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletResponse;
import java.lang.reflect.Method;
import java.util.Map;



@Slf4j
@Aspect
@Component
public class ArtisanLimitAop {
    
    private final Map limitMap = Maps.newConcurrentMap();

    @Around("@annotation(com.artisan.annos.ArtisanLimit)")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        //拿ArtisanLimit的注解
        ArtisanLimit limit = method.getAnnotation(ArtisanLimit.class);
        if (limit != null) {
            //key作用：不同的接口，不同的流量控制
            String key = limit.key();
            RateLimiter rateLimiter = null;
            //验证缓存是否有命中key
            if (!limitMap.containsKey(key)) {
                // 创建令牌桶
                rateLimiter = RateLimiter.create(limit.permitsPerSecond());
                limitMap.put(key, rateLimiter);
                log.info("新建了令牌桶={}，容量={}", key, limit.permitsPerSecond());
            }
            rateLimiter = limitMap.get(key);
            // 拿令牌
            boolean acquire = rateLimiter.tryAcquire(limit.timeout(), limit.timeunit());
            // 拿不到命令，直接返回异常提示
            if (!acquire) {
                log.warn("令牌桶={}，获取令牌失败", key);
                this.responseFail(limit.message());
                return null;
            }
        }
        return joinPoint.proceed();
    }

    
    private void responseFail(String msg) {
        HttpServletResponse response = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();
        ResponseData