常见限流算法介绍（漏桶算法、令牌桶算法）及实现--待整理

 

1.限流
2.限流算法
　　2.1计数器
　　2.2 漏桶算法
　　2.3 令牌桶算法
　　2.4、滑动时间窗
　　2.5、三色速率标记法
三、限流实现
　　3.1 RateLimiter简介(guava的令牌桶实现）
　　3.2 基于 redis 的分布式限流

　　3.3 Spring Cloud GateWay整合redis内置RequestRateLimiter限流应用
　　3.4 AOP + Semaphore限流应用

 

1.限流

每个API接口都是有访问上限的,当访问频率或者并发量超过其承受范围时候,我们就必须考虑限流来保证接口的可用性或者降级可用性.即接口也需要安装上保险丝,以防止非预期的请求对系统压力过大而引起的系统瘫痪.

通常的策略就是拒绝多余的访问,或者让多余的访问排队等待服务,或者引流.

如果要准确的控制QPS,简单的做法是维护一个单位时间内的Counter,如判断单位时间已经过去,则将Counter重置零.此做法被认为没有很好的处理单位时间的边界,比如在前一秒的最后一毫秒里和下一秒的第一毫秒都触发了最大的请求数,将目光移动一下,就看到在两毫秒内发生了两倍的QPS.

2.限流算法

常用的更平滑的限流算法有两种:漏桶算法和令牌桶算法.

两种限流的祖师级算法确有其独到之处，其他实现比如滑动时间窗或者三色速率标记法，其实是“漏桶”与“令牌桶”的变种。要么将“漏桶”容积换成了单位时间，要么是按规则将请求标记颜色进行处理，底层还是“令牌”的思想。

很多传统的服务提供商如华为中兴都有类似的专利,参考: http://www.google.com/patents/CN1536815A?cl=zh

2.1计数器

计数器法是限流算法里最简单也是最容易实现的一种算法。比如我们规定，对于A接口来说，我们1分钟的访问次数不能超过100个。那么我们我们可以设置一个计数器counter，其有效时间为1分钟（即每分钟计数器会被重置为0），每当一个请求过来的时候，counter就加1，如果counter的值大于100，就说明请求数过多；
这个算法虽然简单，但是有一个十分致命的问题，那就是临界问题。

如上图所示，在1:00前一刻到达100个请求，1:00计数器被重置，1:00后一刻又到达100个请求，显然计数器不会超过100，所有请求都不会被拦截；然而这一时间段内请求数已经达到200，远超100。违背定义的固定速率。

2.2 漏桶算法

漏桶(Leaky Bucket)算法思路很简单,水(请求)先进入到漏桶里,漏桶以一定的速度出水(接口有响应速率),当水流入速度过大会直接溢出(访问频率超过接口响应速率),然后就拒绝请求,而当入小于出的情况下，漏桶不起任何作用。可以看出漏桶算法能强行限制数据的传输速率.示意图如下:

      

注意：在我们的应用中，漏桶算法强制限定流量速率后，多出的（溢出的）流量可以被利用起来，并非完全丢弃，我们可以把它收集到一个队列里面，做流量队列，尽量做到合理利用所有资源。

漏桶算法：水（请求）先进入到漏桶里，漏桶以一定的速度出水，当水流入速度过大会直接溢出（拒绝服务），可以看出漏桶算法能强行限制数据的传输速率

流入：以任意速率往桶中放入水滴。
流出：以固定速率从桶中流出水滴。

用白话具体说明：假设漏斗总支持并发100个最大请求，如果超过100个请求，那么会提示系统繁忙，请稍后再试，数据输出那可以设置1个线程池，处理线程数5个，每秒处理20个请求。
缺点：因为当流出速度固定，大规模持续突发量，无法多余处理，浪费网络带宽
优点：无法击垮服务

示例：

可见这里有两个变量,一个是桶的大小,支持流量突发增多时可以存多少的水(burst),另一个是水桶漏洞的大小(rate)，伪代码如下：

double rate;               // leak rate in calls/s
double burst;              // bucket size in calls

long refreshTime;          // time for last water refresh
double water;              // water count at refreshTime

refreshWater() {
    long  now = getTimeOfDay();
    
    //水随着时间流逝,不断流走,最多就流干到0.
    water = max(0, water- (now - refreshTime)*rate); 
    refreshTime = now;
}

bool permissionGranted() {
    refreshWater();
    if (water < burst) { // 水桶还没满,继续加1
        water ++;
        return true;
    } else {
        return false;
    }
}

因为漏桶的漏出速率是固定的参数,所以,即使网络中不存在资源冲突(没有发生拥塞),漏桶算法也不能使流突发(burst)到端口速率.因此,漏桶算法对于存在突发特性的流量来说缺乏效率.

2.3 令牌桶算法

令牌桶算法(Token Bucket)和 Leaky Bucket 效果一样但方向相反的算法,更加容易理解.随着时间流逝,系统会按恒定1/QPS时间间隔(如果QPS=100,则间隔是10ms)往桶里加入Token(想象和漏洞漏水相反,有个水龙头在不断的加水),如果桶已经满了，令牌就溢出了。如果桶未满，令牌可以积累。新请求来临时,会各自拿走一个Token,如果没有Token可拿了就阻塞或者拒绝服务.

令牌桶的另外一个好处是可以方便的改变速度. 一旦需要提高速率,则按需提高放入桶中的令牌的速率. 一般会定时(比如100毫秒)往桶中增加一定数量的令牌, 有些变种算法则实时的计算应该增加的令牌的数量.

令牌桶算法：一个存放固定容量令牌的桶，按照固定速率（每秒/或者可以自定义时间）往桶里添加令牌，然后每次获取一个令牌，当桶里没有令牌可取时，则拒绝服务
令牌桶分为2个动作，动作1(固定速率往桶中存入令牌)、动作2(客户端如果想访问请求，先从桶中获取token)

流入：以固定速率从桶中流入水滴
流出：按照任意速率从桶中流出水滴

技术上使用Google开源工具包Guava提供了限流工具类RateLimiter，该类基于令牌桶算法来完成限流，非常易于使用。RateLimiter是guava提供的基于令牌桶算法的实现类，可以非常简单的完成限流特技，并且根据系统的实际情况来调整生成token的速率。 RateLimiter 是单机（单进程）的限流，是JVM级别的的限流，所有的令牌生成都是在内存中，在分布式环境下不能直接这么用。
优点：支持大的并发，有效利用网络带宽

漏桶和令牌桶的区别：

并不能说明令牌桶一定比漏洞好，她们使用场景不一样。

• 令牌桶算法，放在服务端，用来保护服务端（自己），主要用来对调用者频率进行限流，为的是不让自己被压垮。所以如果自己本身有处理能力的时候，如果流量突发（实际消费能力强于配置的流量限制=桶大小），那么实际处理速率可以超过配置的限制（桶大小）。  
• 而漏桶算法，放在调用方，这是用来保护他人，也就是保护他所调用的系统。主要场景是，当调用的第三方系统本身没有保护机制，或者有流量限制的时候，我们的调用速度不能超过他的限制，由于我们不能更改第三方系统，所以只有在主调方控制。这个时候，即使流量突发，也必须舍弃。因为消费能力是第三方决定的。

2.4、滑动时间窗

滑动窗口，又称rolling window。为了解决这个问题，我们引入了滑动窗口算法。如果学过TCP网络协议的话，那么一定对滑动窗口这个名词不会陌生。下面这张图，很好地解释了滑动窗口算法：

在上图中，整个红色的矩形框表示一个时间窗口，在我们的例子中，一个时间窗口就是一分钟。然后我们将时间窗口进行划分，比如图中，我们就将滑动窗口 划成了6格，所以每格代表的是10秒钟。每过10秒钟，我们的时间窗口就会往右滑动一格。每一个格子都有自己独立的计数器counter，比如当一个请求 在0:35秒的时候到达，那么0:30~0:39对应的counter就会加1。

那么滑动窗口怎么解决刚才的临界问题的呢？我们可以看上图，0:59到达的100个请求会落在灰色的格子中，而1:00到达的请求会落在橘黄色的格 子中。当时间到达1:00时，我们的窗口会往右移动一格，那么此时时间窗口内的总请求数量一共是200个，超过了限定的100个，所以此时能够检测出来触 发了限流。

我再来回顾一下刚才的计数器算法，我们可以发现，计数器算法其实就是滑动窗口算法。只是它没有对时间窗口做进一步地划分，所以只有1格。

由此可见，当滑动窗口的格子划分的越多，那么滑动窗口的滚动就越平滑，限流的统计就会越精确。

2.5、三色速率标记法

三、限流实现

3.1 RateLimiter简介(guava的令牌桶实现）

Google开源工具包Guava提供了限流工具类RateLimiter,该类基于令牌桶算法(Token Bucket)来完成限流,非常易于使用.RateLimiter经常用于限制对一些物理资源或者逻辑资源的访问速率.它支持两种获取permits接口,一种是如果拿不到立刻返回false,一种会阻塞等待一段时间看能不能拿到。原理见《guava--RateLimiter源码分析》

RateLimiter和Java中的信号量(java.util.concurrent.Semaphore)类似,Semaphore通常用于限制并发量.

源码注释中的一个例子,比如我们有很多任务需要执行,但是我们不希望每秒超过两个任务执行,那么我们就可以使用RateLimiter:

final RateLimiter rateLimiter = RateLimiter.create(2.0);
void submitTasks(List<Runnable> tasks, Executor executor) {
    for (Runnable task : tasks) {
        rateLimiter.acquire(); // may wait
        executor.execute(task);
    }
}

另外一个例子,假如我们会产生一个数据流,然后我们想以每秒5kb的速度发送出去.我们可以每获取一个令牌(permit)就发送一个byte的数据,这样我们就可以通过一个每秒5000个令牌的RateLimiter来实现:

final RateLimiter rateLimiter = RateLimiter.create(5000.0);
void submitPacket(byte[] packet) {
    rateLimiter.acquire(packet.length);
    networkService.send(packet);
}

另外,我们也可以使用非阻塞的形式达到降级运行的目的,即使用非阻塞的tryAcquire()方法:

if(limiter.tryAcquire()) { //未请求到limiter则立即返回false
    doSomething();
}else{
    doSomethingElse();
}

 

3.2 基于 redis 的分布式限流

　　单机版中我们了解到 AtomicInteger、RateLimiter、Semaphore 这几种解决方案，但它们也仅仅是单机的解决手段，在集群环境下就透心凉了，后面又讲述了 Nginx 的限流手段，可它又属于网关层面的策略之一，并不能解决所有问题。例如供短信接口，你无法保证消费方是否会做好限流控制，所以自己在应用层实现限流还是很有必要的。

导入依赖

在 pom.xml 中添加上 starter-web、starter-aop、starter-data-redis 的依赖即可，习惯了使用 commons-lang3 和 guava 中的一些工具包…

<dependencies>
    <!-- 默认就内嵌了Tomcat 容器，如需要更换容器也极其简单-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>21.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
</dependencies>

属性配置

在 application.properites 资源文件中添加 redis 相关的配置项

spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=battcn

Limit 注解

创建一个 Limit 注解，不多说注释都给各位写齐全了….

package com.johnfnash.learn.springboot.ratelimiter.annotation;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

// 限流
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Limit {

    /**
     * 资源的名称
     * @return
     */
    String name() default "";
    
    /**
     * 资源的key
     *
     * @return
     */
    String key() default "";

    /**
     * Key的prefix
     *
     * @return
     */
    String prefix() default "";
    
    /**
     * 给定的时间段
     * 单位秒
     *
     * @return
     */
    int period();

    /**
     * 最多的访问限制次数
     *
     * @return
     */
    int count();
    
    /**
     * 类型
     *
     * @return
     */
    LimitType limitType() default LimitType.CUSTOMER;
}

package com.johnfnash.learn.springboot.ratelimiter.annotation;

// 限制的类型
public enum LimitType {

    /**
     * 自定义key
     */
    CUSTOMER,
    /**
     * 根据请求者IP
     */
    IP;
    
}

RedisTemplate

默认情况下 spring-boot-data-redis 为我们提供了StringRedisTemplate 但是满足不了其它类型的转换，所以还是得自己去定义其它类型的模板….

package com.johnfnash.learn.springboot.ratelimiter;

import java.io.Serializable;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisLimiterHelper {

    @Bean
    public RedisTemplate<String, Serializable> limitRedisTemplate(LettuceConnectionFactory factory) {
        RedisTemplate<String, Serializable> template = new RedisTemplate<String, Serializable>();
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        template.setConnectionFactory(factory);
        return template;
    }
    
}

Limit 拦截器（AOP）

熟悉 Redis 的朋友都知道它是线程安全的，我们利用它的特性可以实现分布式锁、分布式限流等组件，限流相比稍微复杂一点，官方虽然没有提供相应的API，但却提供了支持 Lua 脚本的功能，我们可以通过编写 Lua 脚本实现自己的API，同时它是满足原子性的，下面核心就是调用 execute 方法传入我们的 Lua 脚本内容，然后通过返回值判断是否超出我们预期的范围，超出则给出错误提示。

package com.johnfnash.learn.springboot.ratelimiter.aop;

import java.io.Serializable;
import java.lang.reflect.Method;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.lang3.StringUtils;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import com.google.common.collect.ImmutableList;
import com.johnfnash.learn.springboot.ratelimiter.annotation.Limit;
import com.johnfnash.learn.springboot.ratelimiter.annotation.LimitType;

@Aspect
@Configuration
public class LimitInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(LimitInterceptor.class);;
    
    private final String REDIS_SCRIPT = buildLuaScript();
    
    @Autowired
    private RedisTemplate<String, Serializable> redisTemplate;
    
    @Around("execution(public * *(..)) && @annotation(com.johnfnash.learn.springboot.ratelimiter.annotation.Limit)")
    public Object interceptor(ProceedingJoinPoint pjp) {
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();
        Limit limitAnno = method.getAnnotation(Limit.class);
        LimitType limitType = limitAnno.limitType();
        String name = limitAnno.name();
        
        String key = null;
        int limitPeriod = limitAnno.period();
        int limitCount = limitAnno.count();
        switch (limitType) {
        case IP:
            key = getIpAddress();
            break;
        case CUSTOMER:
            // TODO 如果此处想根据表达式或者一些规则生成 请看 一起来学Spring Boot | 第二十三篇：轻松搞定重复提交（分布式锁）
            key = limitAnno.key();
            break;
        default:
            break;
        }
        
        ImmutableList<String> keys = ImmutableList.of(StringUtils.join(limitAnno.prefix(), key));
        try {
            RedisScript<Number> redisScript = new DefaultRedisScript<Number>(REDIS_SCRIPT, Number.class);
            Number count = redisTemplate.execute(redisScript, keys, limitCount, limitPeriod);
            logger.info("Access try count is {} for name={} and key = {}", count, name, key);
            if(count != null && count.intValue() <= limitCount) {
                return pjp.proceed();
            } else {
                throw new RuntimeException("You have been dragged into the blacklist");
            }
        } catch (Throwable e) {
            if (e instanceof RuntimeException) {
                throw new RuntimeException(e.getLocalizedMessage());
            }
            throw new RuntimeException("server exception");
        }
    }
    
    /**
     * 限流 脚本
     *
     * @return lua脚本
     */
    private String buildLuaScript() {
        StringBuilder lua = new StringBuilder();
        lua.append("local c")
           .append("
c = redis.call('get', KEYS[1])")
           // 调用不超过最大值，则直接返回
           .append("
if c and tonumber(c) > tonumber(ARGV[1]) then")
           .append("
return c;")
           .append("
end")
           // 执行计算器自加
           .append("
c = redis.call('incr', KEYS[1])")
           .append("
if tonumber(c) == 1 then")
           // 从第一次调用开始限流，设置对应键值的过期
           .append("
redis.call('expire', KEYS[1], ARGV[2])")
           .append("
end")
           .append("
return c;");
        return lua.toString();
    }
    
    private static final String UNKNOWN = "unknown";

    public String getIpAddress() {
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        String ip = request.getHeader("x-forwarded-for");
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
            ip = request.getHeader("Proxy-Client-IP");
        }
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
            ip = request.getHeader("WL-Proxy-Client-IP");
        }
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
            ip = request.getRemoteAddr();
        }
        return ip;
    }
    
}

控制层

在接口上添加 @Limit() 注解，如下代码会在 Redis 中生成过期时间为 100s 的 key = test 的记录，特意定义了一个 AtomicInteger 用作测试：

package com.johnfnash.learn.springboot.ratelimiter.controller;

import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import com.johnfnash.learn.springboot.ratelimiter.annotation.Limit;

@RestController
public class LimiterController {

    private static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger();

    @Limit(key = "test", period = 100, count = 10)
    // 意味著 100S 内最多允許訪問10次
    @GetMapping("/test")
    public int testLimiter() {
        return ATOMIC_INTEGER.incrementAndGet();
    }
    
}

测试

完成准备事项后，启动 启动类 自行测试即可，测试手段相信大伙都不陌生了，如 浏览器、postman、junit、swagger，此处基于 postman，未达设定的阀值时，正常响应

 注：上面的方式是使用计数器的限流方式，无法处理临界的时候，大量请求的的情况。要解决这个问题，可以使用redis中列表类型的键来记录最近N次访问的时间，一旦键中的元素超过N个，就判断时间最早的元素距现在的时间是否小于M秒。如果是则表示用户最近M秒的访问次数超过了N次；如果不是就将现在的时间加入列表中同时把最早的元素删除（可以通过脚本功能避免竞态条件）。由于需要记录每次访问的时间，所以当要限制“M时间最多访问N次”时，如果“N”的数值较大，此方法会占用较多的存储空间，实际使用时还需要开发者自己去权衡。

下面的解决思路的实现如下：

    /**
     * 限流 脚本（处理临界时间大量请求的情况）
     *
     * @return lua脚本
     */
    private String buildLuaScript2() {
        StringBuilder lua = new StringBuilder();
        lua.append("local listLen, time")
           .append("
listLen = redis.call('LLEN', KEYS[1])")
           // 不超过最大值，则直接写入时间
           .append("
if listLen and tonumber(listLen) < tonumber(ARGV[1]) then")
                .append("
local a = redis.call('TIME');")
                .append("
redis.call('LPUSH', KEYS[1], a[1]*1000000+a[2])")
           .append("
else")
               // 取出现存的最早的那个时间，和当前时间比较，看是小于时间间隔
               .append("
time = redis.call('LINDEX', KEYS[1], -1)")
               .append("
local a = redis.call('TIME');")
               .append("
if a[1]*1000000+a[2] - time < tonumber(ARGV[2])*1000000 then")
                   // 访问频率超过了限制，返回0表示失败
                   .append("
return 0;")
               .append("
else")                   
                   .append("
redis.call('LPUSH', KEYS[1], a[1]*1000000+a[2])")
                   .append("
redis.call('LTRIM', KEYS[1], 0, tonumber(ARGV[1])-1)")
               .append("
end")   
           .append("
end")
           .append("
return 1;");
        return lua.toString();
    }

调用地方修改如下：

if(count != null && count.intValue() == 1) {
    return pjp.proceed();
} else {
    throw new RuntimeException("You have been dragged into the blacklist");
}

补充，最近执行 buildLuaScript2(） 中的lua脚本，报错

Write commands not allowed after non deterministic commands.

这个错误的原因大家可以参见这篇文章，大致原因跟redis集群的重放和备份策略有关，相当于我调用TIME操作，会在主从各执行一次，得到的结果肯定会存在差异，这个差异就给最终逻辑正确性带来了不确定性。在redis 4.0之后引入了redis.replicate_commands()来放开限制。

于是，在 buildLuaScript2 的 lua 脚本最前面加上 “redis.replicate_commands();”，错误得以解决。

3.3 Spring Cloud GateWay整合redis内置RequestRateLimiter限流应用

3.4 AOP + Semaphore限流应用

见《AOP+Semaphore实现单节点的接口(方法)限流》

参考: https://github.com/springside/springside4/wiki/Rate-Limiter

https://en.wikipedia.org/wiki/Token_bucket

https://en.wikipedia.org/wiki/Leaky_bucket

https://blog.csdn.net/johnf_nash/article/details/89791808