《亿级流量网站架构核心技术》-高可用-限流详解

背景

项目中遇到一个场景，用户点击按钮给缺勤员工发送扣薪通知。按照需求是一个月只会发送一次通知，在代码逻辑中已经做了兜底策略--在发送邮件之前先判断本月通知是否已发送，发送完之后将发送状态进行持久化。发邮件的时间大概在1分钟之内。

本来觉得逻辑很完善了，可当用户使用时会连续点击按钮，就会出现类似多并发或者重复提交的场景。开始设计了几个思路。

加锁，由于服务部署在两个服务上，所以需要加分布式锁，之前也是遇到类似的问题也是这样搞的。
加防止重复提交校验，时间窗内不允许重复提交。但重复提交更多的是用于表单提交，其实这也也是一个思路。
限流，也是这篇文字介绍的内容，其主要代码贴在了分布式限流。通过设置时间窗口，和限流阈值来限制短时间内多次访问接口。

可能限流并不是所遇场景最优解决方案，也是为了学习吧。

以下内容整理自《亿级流量网站架构核心技术》

限流的目的是通过对并发访问/请求进行限速或者一个时间窗口内的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务，如定向到错误页面。也可以进行排队/等待，降级。

高并发常见的限流有：限制总并发数，限制时间窗口内的平均速率

限流算法

限流算法主要有两种：

平均速率限流：令牌桶算法，漏桶算法，

总数量限流：简单粗暴的计数器

平均速率限流:

令牌桶算法：是一个存放固定容量令牌的桶，按照固定速率往桶里添加令牌。具体描述如下：

按照500毫秒的固定速率往桶中添加令牌，流入速率可以调整。
桶中最多存放b个令牌，当桶满时，就将新增的令牌丢弃
当有请求进来的时候，从桶中获取令牌
若桶中没有令牌，那么该请求就会被限流（丢弃或者缓冲区等待）

漏桶算法：请求如同滴入漏桶中的水，漏桶的容量是固定的，通过控制流出的速率来进行限流。

一个固定容量的漏桶，按照常量固定速率流出水滴。
如果桶是空的，则不需流出水滴.可以以任意速率流入水滴到漏桶。
如果流入水滴超出了桶的容量，则流入的水滴溢出了被丢弃，而漏桶容量是不变的。

令牌桶是按照固定速率往桶中添加令牌，请求是否被处理需要看桶中令牌是否足够，当令牌数减为零时，则拒绝新的请求。

漏桶则是按照常量固定速率流出请求，流入请求速率任意，当流入的请求数累积到漏桶容量时，则新流入的请求被拒绝。

令牌桶限制的是平均流入速率，允许突发请求，只要有令牌就可以处理，并允许一定程度的突发流量。

漏桶限制的是常量流出速率 (即流出速率是一个固定常量值，通过逐步扩大流出速率，从而平滑突发流入速率。

令牌桶允许一定程度的突发，而漏桶主要目的是平滑流入速率。

总数量限流

计数器限流，主要用来限制总并发数，比如数据库连接池大小、线程池大小、秒杀并发数都是计数器的用法。只要全局总请求数或者一定时间段的总请求数达到设定阙值，则进行限流。

单体应用限流

对于单体应用而言，限流方式从整体和具体接口两方面进行考虑。即使拓展到分布式系统，对于每台机器部署的应用来说，以下方式依然具有参考意义。

整体层面

从整体来考虑限流，可以从总的并发数和总的资源考虑。

限流总的并发数

对于应用系统来说，一定会有极限并发/请求数，即总有一个 TPS/QPS 阙值。一旦超了阙值，则系统就会不响应用户请求或响应得非常慢。因此需要最好根据需求设计QPS阈值，进行过载保护，以防止大量请求涌入击垮系统。

限流总的资源数

同样对于数据库连接、线程这样的资源而言，也需要加以限制。可以使用池化技术来限制总资源数，如连接池、线程池。假设分配给每个应用的数据库连接是 100，那么本应用最多可以使用 100 个资源，超出则可以等待或者抛异常

具体接口

对于具体接口限流的方式主要有以下几种

限流某个接口的并发数

如抢购接口，可能会有突发访问情况，若访问量太大则容易造成系统崩溃，这个时候就需要限制这个接口的总并发/请求数总请求数了。要么让用户排队，要么告诉用户没货了

限流某个接口的时间窗请求数

对于一些被其他系统调用的基础服务，可能请求并发量不是很多，但是频繁的进行大规模的查询也可能将基础服务搞崩溃。这时候就需要根据一个时间窗口内来时限制某个接口每秒/每分钟的请求数调用量，对每秒/每分钟的调用量进行限速。

平滑限流某个接口的请求数

上面的限流方式都不能很好地应对突发请求，即瞬间请求可能都被允许，从而导致些问题。因此，在一些场景中需要对突发请求进行平滑限流，降低平均速率请求，比每隔 200 毫秒处理一个请求。这个时候用到上面提到的两种算法:令牌桶和漏桶算法。

分布式限流

分布式限流最关键的是要将限流服务做成原子化，而解决方案可以使用 Redis+Lua或者 Nginx+Lua 技术进行实现，通过这两种技术可以实现高并发和高性能。

如下操作因是在一个 Lua 脚本中，又因 Redis 是单线程模型，因此线程安全。

    @Bean
    public DefaultRedisScript<Long> limitScript() {
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(limitScriptText());
        redisScript.setResultType(Long.class);
        return redisScript;
    }

    /**
     * 限流脚本
     */
    private String limitScriptText(){
                //  限流key
        return "local key = KEYS[1]\n" +   
                // 限流阈值
                "local count = tonumber(ARGV[1])\n" +
                // 限流时间窗
                "local time = tonumber(ARGV[2])\n" +
                // 获取限流key的缓存值，
                "local current = redis.call('get', key);\n" +
                // 若缓存值大于阈值，则返回缓存值
                "if current and tonumber(current) > count then\n" +
                "    return current;\n" +
                "end\n" +
                // 否则，缓存值加1，并加入缓存
                "current = redis.call('incr', key)\n" +
                // 如果当前值==1，即初始化的时候，设置过期时间
                "if tonumber(current) == 1 then\n" +
                "    redis.call('expire', key, time)\n" +
                "end\n" +
                "return current;";
    }

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter{
    // 限流key
    public String key() default "key";
  
    // 限流时间,单位秒
    public int time() default 60;

     //限流次数
    public int count() default 100;
}

@Aspect
@Component
public class RateLimiterAspect{
    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    @Autowired
    private RedisScript<Long> limitScript;


    // 配置织入点
    @Pointcut("@annotation(RateLimiter)")
    public void rateLimiterPointCut(){}

    @Before("rateLimiterPointCut()")
    public void doBefore(JoinPoint point) throws Throwable {
        Signature signature = joinPoint.getSignature();
        MethodSignature methodSignature = (MethodSignature) signature;
        Method method = methodSignature.getMethod();
        if (method != null){
            return;
        }
        RateLimiter rateLimiter = method.getAnnotation(RateLimiter.class);
        if(Objects.isNull(rateLimiter)){
          return;
        }
        String key = rateLimiter.key();
        int time = rateLimiter.time();
        int count = rateLimiter.count();
      
        List<Object> keys = Collections.singletonList(key);
        // 获取缓存窗口中该key的请求数
        Long number = redisTemplate.execute(limitScript, keys, count, time);
        if (StringUtils.isNull(number) || number.intValue() > count){
            throw new ServiceException("访问过于频繁，请稍后再试");
        }
    }