Spring Boot高可用限流三种实现解决方案

/**
 * 固定窗口时间算法
 * @return
*/
public class Counter {
  public long timeStamp = System.currentTimeMillis(); // 当前时间
  public int reqCount = 0; // 初始化计数器
  public final int limit = 1000; // 时间窗口内最大请求数
  public final long interval = 1000 * 60; // 时间窗口ms
​
  public boolean limit() {
    long now = System.currentTimeMillis();
    if (now < timeStamp + interval) {
      // 在时间窗口内
      reqCount++;
      // 判断当前时间窗口内是否超过最大请求控制数
      return reqCount <= limit;
    } else {
      timeStamp = now;
      // 超时后重置
      reqCount = 1;
      return true;
    }
  }
}

 /**
     * 单位时间划分的小周期（单位时间是1分钟，10s一个小格子窗口，一共6个格子）
     */
    private int SUB_CYCLE = 10;
    /**
     * 每分钟限流请求数
     */
    private int thresholdPerMin = 100;
    /**
     * 计数器, k-为当前窗口的开始时间值秒，value为当前窗口的计数
     */
    private final TreeMap<Long, Integer> counters = new TreeMap<>();
   /**
     * 滑动窗口时间算法实现
     */
    boolean slidingWindowsTryAcquire() {
        long currentWindowTime = LocalDateTime.now().toEpochSecond(ZoneOffset.UTC) / SUB_CYCLE * SUB_CYCLE; //获取当前时间在哪个小周期窗口
        int currentWindowNum = countCurrentWindow(currentWindowTime); //当前窗口总请求数
        //超过阀值限流
        if (currentWindowNum >= thresholdPerMin) {
            return false;
        }
        //计数器+1
        counters.get(currentWindowTime)++;
        return true;
    }
   /**
    * 统计当前窗口的请求数
    */
    private int countCurrentWindow(long currentWindowTime) {
        //计算窗口开始位置
        long startTime = currentWindowTime - SUB_CYCLE* (60s/SUB_CYCLE-1);
        int count = 0;
        //遍历存储的计数器
        Iterator<Map.Entry<Long, Integer>> iterator = counters.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Long, Integer> entry = iterator.next();
            // 删除无效过期的子窗口计数器
            if (entry.getKey() < startTime) {
                iterator.remove();
            } else {
                //累加当前窗口的所有计数器之和
                count =count + entry.getValue();
            }
        }
        return count;
    }

     <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>29.0-jre</version>
        </dependency>

/**
 * @author fjzheng
 * @version 1.0
 * @date 2021/9/1 10:38
 */
@Slf4j
@RestController
@RequestMapping("/test")
@ResponseResultBody
public class TestController {
    /**
     * 限流策略 ：1秒钟2个请求
     */
    private final RateLimiter limiter = RateLimiter.create(2.0);
    private DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    @GetMapping("/rateLimit")
    public String testLimiter() {
        //500毫秒内，没拿到令牌，就直接进入服务降级
        boolean tryAcquire = limiter.tryAcquire(50, TimeUnit.MILLISECONDS);
        if (!tryAcquire) {
            log.warn("进入服务降级，时间{}", LocalDateTime.now().format(dtf));
            return "当前排队人数较多，请稍后再试！";
        }
​
        log.info("获取令牌成功，时间{}", LocalDateTime.now().format(dtf));
        return "请求成功";
    }
​
}

/**
 * @author fjzheng
 * @version 1.0
 * @date 2021/10/25 00:06
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.TYPE})
@Documented
public @interface RateLimit {
    /**
     * 资源的key,唯一
     * 作用：不同的接口，不同的流量控制
     */
    String key() default "";
​
    /**
     * 最多的访问限制次数
     */
    double permitsPerSecond () ;
​
    /**
     * 获取令牌最大等待时间
     */
    long timeout();
​
    /**
     * 获取令牌最大等待时间,单位(例:分钟/秒/毫秒) 默认:毫秒
     */
    TimeUnit timeunit() default TimeUnit.MILLISECONDS;
​
    /**
     * 得不到令牌的提示语
     */
    String msg() default "系统繁忙,请稍后再试.";
​
}

/**
 * @author fjzheng
 * @version 1.0
 * @date 2021/10/25 00:10
 */
@Slf4j
@Component
@Aspect
public class RateLimitAop {
    /**
     * 不同的接口，不同的流量控制
     * map的key为 Limiter.key
     */
    private final Map<String, RateLimiter> limitMap = Maps.newConcurrentMap();
​
    @Around("@annotation(com.shepherd.mall.seckill.annotation.RateLimit)")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable{
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        //拿limit的注解
        RateLimit limit = method.getAnnotation(RateLimit.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.debug("令牌桶={}，获取令牌失败",key);
                this.responseFail(limit.msg());
                return null;
            }
        }
        return joinPoint.proceed();
    }
​
    /**
     * 直接向前端抛出异常
     * @param msg 提示信息
     */
    private void responseFail(String msg)  {
        HttpServletResponse response=((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();
        ResponseVO<Object> responseVO = ResponseVO.failure(400, msg);
        WebUtil.writeJson(response, responseVO);
    }
}

/**
 * @author fjzheng
 * @version 1.0
 * @date 2021/9/1 10:38
 */
@Slf4j
@RestController
@RequestMapping("/test")
@ResponseResultBody
public class TestController {
    @GetMapping("/limit2")
    @RateLimit(key = "limit2", permitsPerSecond = 1, timeout = 50, timeunit = TimeUnit.MILLISECONDS, msg = "当前排队人数较多，请稍后再试！")
    public String limit2() {
        log.info("令牌桶limit2获取令牌成功");
        return "ok";
    }
}

local key = "rate.limit:" .. KEYS[1]
local limit = tonumber(ARGV[1])
local expire_time = ARGV[2]
​
local is_exists = redis.call("EXISTS", key)
if is_exists == 1 then
    if redis.call("INCR", key) > limit then
        return 0
    else
        return 1
    end
else
    redis.call("SET", key, 1)
    redis.call("EXPIRE", key, expire_time)
    return 1
end

local locks = require "resty.lock"
​
local function acquire()
    local lock =locks:new("locks")
    local elapsed, err =lock:lock("limit_key") --互斥锁 保证原子特性
    local limit_counter =ngx.shared.limit_counter --计数器
​
    local key = "ip:" ..os.time()
    local limit = 5 --限流大小
    local current =limit_counter:get(key)
​
    if current ~= nil and current + 1> limit then --如果超出限流大小
       lock:unlock()
       return 0
    end
    if current == nil then
       limit_counter:set(key, 1, 1) --第一次需要设置过期时间，设置key的值为1，
--过期时间为1秒
    else
        limit_counter:incr(key, 1) --第二次开始加1即可
    end
    lock:unlock()
    return 1
end
ngx.print(acquire())

http {
    include       mime.types;
    default_type  application/octet-stream;
​
    #cache
    lua_shared_dict dis_cache 128m;
​
    #限流设置
    limit_req_zone $binary_remote_addr zone=contentRateLimit:10m rate=2r/s;
​
    #根据IP地址来限制，存储内存大小10M
    limit_conn_zone $binary_remote_addr zone=addr:1m;
​
    sendfile        on;
    #tcp_nopush     on;
​
    #keepalive_timeout  0;
    keepalive_timeout  65;
​
    #gzip  on;
​
    server {
        listen       80;
        server_name  localhost;
        location /brand {
            limit_conn addr 2;
            proxy_pass http://192.168.211.1:18081;
        }
​
        location /update_content {
            content_by_lua_file /root/lua/update_content.lua;
        }
​
        location /read_content {
            limit_req zone=contentRateLimit burst=4 nodelay;
            content_by_lua_file /root/lua/read_content.lua;
        }
    }
}

user  root root;
worker_processes  1;
​
events {
    worker_connections  1024;
}
​
http {
    include       mime.types;
    default_type  application/octet-stream;
​
    #cache
    lua_shared_dict dis_cache 128m;
​
    #限流设置
    limit_req_zone $binary_remote_addr zone=contentRateLimit:10m rate=2r/s;
​
    sendfile        on;
    #tcp_nopush     on;
​
    #keepalive_timeout  0;
    keepalive_timeout  65;
​
    #gzip  on;
​
    server {
        listen       80;
        server_name  localhost;
​
        location /update_content {
            content_by_lua_file /root/lua/update_content.lua;
        }
​
        location /read_content {
            limit_req zone=contentRateLimit burst=4 nodelay;
            content_by_lua_file /root/lua/read_content.lua;
        }
    }
}