微服务架构-流量管控-一一网

1. 流控定义与使用场景

在日常生活中随处可见流量管控的场景，如车流量、航空流量、人流量的管控，都是为了更好地保护通路（资源）的运行能力，避免通路过载造成通路不可用。最明显的例子就是每年国庆/五一的高速堵车，遇到节假日车流量是平常的数十倍之多，高速路承载能力是是有限的，那时就可以看到四处堵车的壮观景象。当时看到交警蜀黍在高速路上设了很多路卡来控制车流量过大的问题，虽然还是很堵车但是如果不去管控，就会造成“大家挤，谁都别想走“的窘境。同样，类比到我们的软件系统中，比如要通行的车辆就是需要处理的请求，道路就是我们构建的系统的通信请求链路，那么这些路卡就是流控的组件。

流控的定义: 保护下游服务的有限资源不会被流量冲垮，用于保护服务集群的可用性，流控的精确度取决于采样的时间片，如果在单位时间内采集越多的请求数据则统计结果越精确，流控也就越精确。
流控使用场景: 可以将整个系统看作一个巨大的流量漏斗，用户层的请求进来后，在目前微服务架构下，由用户层、网关层、业务逻辑层、数据访问层、基础设施层等组成，可用于网关开放接口的调用频率次数限制，微服务调用的频率限制，消费消息削峰填谷。

2. 流控算法

2.1 时间窗口

基于一个给定的时间窗口，维护一个计数器用于统计访问次数，每次有请求时都会先去判断是否到达时间窗口的结束时间，如果到达则更新时间窗口开始时间戳，然后再去判断是否超过请求数的阀值，主要实现代码如下。

public class SimpleWindowThrottler {
    //时间窗口大小
    private long windowSize = 20;
    //请求阀值
    private long threshold = 200;
    //请求计数
    private long counter = 0L;
    //时间窗口的开始时间戳
    private long lastReqTime = System.currentTimeMillis();

    public boolean tryQuire() {
        long now = System.currentTimeMillis();
        //超过了上一次请求时间开始的时间窗口，重置时间窗口与计数器
        if (now - lastReqTime > windowSize) {
            counter = 0L;
            lastReqTime = now;
        }
        //小于阀值则请求数+1，并返回true，可继续访问
        if (counter < threshold) {
            counter++;
            return true;
        }
        //超过了阀值则false，不可继续访问
        return false;
    }
}
复制代码

显而易见，这种简单时间窗口在两个时间窗口交替时会造成一些问题，譬如5秒内100次的限制，刚好4.99秒的通过请求数是0，5.0秒的请求数是100，5.01秒的请求数是100，这样就会产生很多请求毛刺，也可理解为临界问题。

为了解决上述的问题，滑动窗口就是一个改进实现，核心原理就是将时间窗口再细分，然后每个子窗口维护一个独立的计数器用于统计访问量，到时时间窗口向右滑动一格。

当请求进来时，先判断当前请求属于这5个1秒的时间窗口中的哪个，然后将对应的时间窗口对应的统计值加1，再判断当前加上前4个时间窗口的次数总和是否已经超过了设置的阈值。当时间已经到达第6个时间窗口时，清除第一个时间窗口的统计。随着时间的推移，统计值就随着各个时间片的滚动，不断地进行统计。

public class SlidingWindowThrottler {
    
    //循环队列，就是装多个窗口用，该数量是windowSize的2倍
    private AtomicInteger[] timeSlices;
    //队列的总长度
    private int timeSliceSize;
    //每个时间片的时长，以毫秒为单位
    private int timeMillisPerSlice;
    //共有多少个时间片（即窗口长度）
    private int windowSize;
    //在一个完整窗口期内允许通过的最大阈值
    private int threshold;
    //该滑窗的起始创建时间，也就是第一个数据
    private long beginTimestamp;
    //最后一个数据的时间戳
    private long lastAddTimestamp;
    //1秒一个时间片，共4个，请求阀值8
    SlidingWindow window = new SlidingWindow(100, 4, 8); 

    public void tryQuire() {
        return window.addCount(2);
    }

    public SlidingWindow(int timeMillisPerSlice, int windowSize, int threshold) {
        this.timeMillisPerSlice = timeMillisPerSlice;
        this.windowSize = windowSize;
        this.threshold = threshold;
        this.timeSliceSize = windowSize * 2;
        reset();
    }
 
    /**
     * 初始化
     */
    private void reset() {
        beginTimestamp = SystemClock.now();
        //窗口个数
        AtomicInteger[] localTimeSlices = new AtomicInteger[timeSliceSize];
        for (int i = 0; i < timeSliceSize; i++) {
            localTimeSlices[i] = new AtomicInteger(0);
        }
        timeSlices = localTimeSlices;
    }
 
    /**
     * 计算当前所在的时间片的位置
     */
    private int locationIndex() {
        long now = SystemClock.now();
        //如果当前的key已经超出一整个时间片了，那么就直接初始化就行了，不用去计算了
        if (now - lastAddTimestamp > timeMillisPerSlice * windowSize) {
            reset();
        }
        return (int) (((now - beginTimestamp) / timeMillisPerSlice) % 
                        timeSliceSize);
    }
 
    /**
     * 增加count个数量
     */
    public boolean addCount(int count) {
        //当前自己所在的位置，是哪个小时间窗
        int index = locationIndex();

        //清空前面的windowSize到2*windowSize之间的数据格的数据
        //譬如1秒分4个窗口，那么数组共计8个窗口
        //当前index为5时，就清空6、7、8、1。然后把2、3、4、5的加起来就是该窗口内的总和
        for (int i = 1; i <= windowSize; i++) {
            int j = index + i;
            if (j >= windowSize * 2) {
                j -= windowSize * 2;
            }
            timeSlices[j].set(0);
        }     
        //把窗口内的所有时间窗口下的请求数相加
        int sum = 0;
        sum += timeSlices[index].addAndGet(count);
        for (int i = 1; i < windowSize; i++) {
            sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();
        }

        lastAddTimestamp = SystemClock.now();
        return sum >= threshold;
    }
}
复制代码

2.2 漏桶

在有些应用场景中，业务希望不是一下子切断请求流量，而是将请求流量控制在一定的速度内，这时滑动窗口就无法很好地解决请求平滑度问题，漏桶是基于流速来进行流量控制而达到平整流量的效果。

漏桶算法中可以把请求看作是往桶里注水，桶底漏出的水代表离开缓冲区被服务器处理的请求，桶口溢出的水代表被丢弃的请求，以下为其中的概念:

最大允许请求数N：桶的大小;
时间窗口大小T：一整桶水漏完的时间;
最大访问速率V：一整桶水漏完的速度，即N/T;
请求被限流：桶注水的速度比漏水的速度快，最终导致桶内水溢出.

假设起始时刻桶是空的，每次访问都会往桶里注入一单位体积的水量，那么当我们以小于等于N/T的速度往桶里注水时, 桶内的水就永远不会溢出。反之，一旦实际注水速度超过漏水速度, 桶里就会产生越来越多的积水，直到溢出为止。同时漏水的速度永远被控制在N/T以内，这就实现了平滑流量的目的。

** 图.jpg
**

<图片来自于网络>

public class BarrelThrottler {
    //桶的容量
    private long capacity;
    //漏水的速度，即 capacity / duration
    private double velocity;
    //一桶水漏完的时间
    private long duration;
    //桶内剩余的水量
    private long left;
    //上次注水的时间
    private long lastInjectTime;

    public boolean tryQuire() {
        long now = System.currentTimeMillis();
        // 过去这段时间内漏掉的水量 = (当前时间-上次注水时间) * 漏水速度
        // 如果当前时间相比上次注水时间相隔太久，时间差足够大， 则桶内的剩余水量就是0
        // 当前剩余的水 = 之前的剩余水量 - 过去这段时间内漏掉的水量
        left = Math.max(0, left - (long)((now - lastInjectTime) * velocity));
        // 往当前水量基础上注一单位水，只要没有溢出就代表可以访问
        if (left + 1 <= capacity) {
            lastInjectTime = now;
            left++;
            return true;
        }
        return false;
    }
}
复制代码

漏桶的优势在于能够平滑流量，如果流量不是均匀的，那么漏桶算法与滑动窗口算法一样无法做到真正的精确控制。极端情况下，漏桶在时间窗口 T 内也会放进相当于2倍阈值N的流量。如果访问量相比窗口大小N大很多，在窗口（0~T）一开始的0时刻就直接涌进来，使得漏桶在时间t时刻溢出，然后在剩余的T – t时间内按照N / T的速度流入，在总的窗口内的访问量，只要t够小，访问量就接近2N。

2.3 令牌桶

令牌桶算法的原理是系统以恒定的速率产生令牌，然后把令牌放到令牌桶中，令牌桶有一个容量，当令牌桶满了的时候，再加入令牌时则会丢弃；当想要处理一个请求的时候，需要从令牌桶中取出一个令牌，如果此时令牌桶中没有令牌，那么则拒绝该请求。

**
**

<图片来自于网络>

public class TokenThrotter {

    //阻塞队列
    private ArrayBlockingQueue<String> blockingQueue;
    //令牌个数
    private int limit;
    //放入令牌的速率
    private int period;

    public TokenLimiter(int limit, int period) {
        this.limit = limit;
        this.period = period;
        blockingQueue = new ArrayBlockingQueue<>(limit);
        init();
        start();
    }

    private void init() {
        for (int i = 0; i < limit; i++) {
            blockingQueue.add("1");
        }
    }

    public boolean tryQuire() {
        return blockingQueue.poll() == null;
    }

    public void addToken() {
        blockingQueue.add("1");
    }

    //开启定时任务线程，每隔10s放入
    public void start() {
        newScheduledThreadPool(1).schedule(() -> addToken(), 
            10, period, TimeUnit.SECONDS);
    }
}
复制代码

生产环境中使用令牌桶的话，可以考虑借助Guava中提供的RateLimiter。它的实现是多线程安全的，调用RateLimiter#acquire时，如果剩余令牌不足，会阻塞线程一段时间直至有足够的可用令牌。除去默认的SmoothBursty策略外，RateLimiter提供SmoothWarmingUp的策略，支持设置一个热身期，热身期内，RateLimiter会平滑地将放令牌的速率加大，直致最大速率。上面讲述了几种主流的流控算法，可以做下技术选型对比。

3.流控的实战经验

3.1 分布式流控

在实际生产环境中，应用服务通常是分布式部署的，如果共用的资源（例如数据库）或者依赖的下游服务都需要流量限制，那么就需要分布式流控。分布式环境下做流控的核心算法思路与单机流控基本一致，区别在于需要实现一种同步机制来保证全局配置，同步机制的实现可以有中心化和去中心化两种思路：

中心化：配置由一个中心系统统一管控，应用进程通过向中心系统申请的方式获取流控配置。状态的一致性在中心系统维护，实现简单；中心系统节点的不可用会导致流控出错，需要有额外的保护;
去中心化：应用进程独立保存和维护流控配置状态，集群内周期性异步通讯以保持状态一致。相比中心化方案，去中心化方案能够降低中心化单点可靠性带来的影响，但实现上比较复杂，状态的一致性难以保证.

3.2 实战场景

梳理出流量场景是非常重要的事情，主要梳理出核心服务接口的流量，并针对这些流量采用的规则与措施。

3.2.1 Nginx负载均衡流控

在负载均衡层就可以做流量管控，当时使用Nginx，其作为一款具有良好扩展性的负载均衡中间件，可支持多种插件，其中一个ngx_http_limit_req_module的可选模块用于流量管控，底层使用的是漏桶算法。

limit_req_zone $binary_remote_addr zone=mylimit:10m rate=50r/s;

server {
    location /login/ {
        limit_req zone=mylimit;
        proxy_pass http://my_upstream;
    }
}
复制代码

3.2.2 基于redis实现流控

在网关层经常要防止恶意攻击，由同一个ip发送大量的请求到网关层，为了过滤这些攻击请求可使用的redis简单的时间窗口来限制请求的频率，要求简单粗暴，核心代码实现如下。

FUNCTION LIMIT_API_CALL(ip)
ts = CURRENT_UNIX_TIME()
keyname = ip+":"+ts
current = GET(keyname)
IF current != NULL AND current > 10 THEN
    ERROR "too many requests per second"
ELSE
    MULTI
        INCR(keyname,1)
        EXPIRE(keyname,10)
    EXEC
    PERFORM_API_CALL()
END
复制代码

基于redis实现用户认证限流的过程中，业务需要放缓用户登录验证的请求频率，而不是拒绝不处理这些登录验证，这时可以使用令牌桶/漏桶的方式平滑处理这些登录请求。

存储的数据结构: <key:value> 
示例:
{
    xxxxxxxx:
    {
       ”token" : "xxxxx",
       "timestamp" xxxx
    }
}
------------------------------------------
local tokens_key = KEYS[1]
local timestamp_key = KEYS[2]

local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local fill_time = capacity/rate
local ttl = math.floor(fill_time*2)

local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
  last_tokens = capacity
end

local last_refreshed = tonumber(redis.call("get", timestamp_key))
if last_refreshed == nil then
  last_refreshed = 0
end

local delta = math.max(0, now-last_refreshed)
local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
local allowed = filled_tokens >= requested
local new_tokens = filled_tokens
if allowed then
  new_tokens = filled_tokens - requested
end

redis.call("setex", tokens_key, ttl, new_tokens)
redis.call("setex", timestamp_key, ttl, now)

return { allowed, new_tokens }
复制代码

基于redis的zset数据架构可以较简单实现滑动时间窗口的限流功能，主要代码如下所示，这里使用的jedis客户端在redisson客户端封装中已有这些限流接口可方便使用，可用于分布式服务调用限流的场景。

long now = System.currentTimeMillis();
long maxScoreMs = now - windowInSecond * 1000;

Transaction redis = jedisPool.getResource().multi();
redis.zremrangeByScore(key, 0, maxScoreMs);
redis.zadd(key, now, now + "-" + Math.random()); // 加入一个随机值使得member不重复
redis.expire(key, windowInSecond);
redis.exec();
复制代码

3.2.3 Sentinel限流

在微服务中会经常使用到Sentinel组件，Sentinel是一款基于滑动时间窗口实现的高性能的限流组件，有限流、降级、整流等多种功能，而且也开发了针对Spring/Dubbo/Spring cloud等框架的使用套件，非常方便接入，而且还支持集群与中心化配置来使用，Sentinel核心思想也是基于滑动窗口设计，下图是限流的核心过程。