五.Go微服务–自适应限流

1. 前言

令牌桶与 漏桶 两种算法最大的一个问题就是他们都属于需要提前设置阈值的算法，基于 QPS 进行限流的时候最麻烦的就是这个阈值应该怎么设定。一般来说我们可以通过压测来决定这个阈值。但是也会存在问题

1. 如果每个系统上线前都要经过很严格的压测，那么成本相对来说会比较大
2. 很多时候压测都会在测试环境进行压测，测试环境一般来说和生产环境会有一定的差异，即使我们在生产环境做了压测，现在我们的应用都是以容器的形式跑在不同的宿主机上的，每台宿主机上的差异，以及不同的负载都会导致这个压测时的结果不一定就一定是正确的
3. 当机器型号，数量等发生改变时，之前的压测的指标能不能用其实是一个问题， 这些数据对于系统的负载的影响其实不是线性的。 举个例子，之前一台机子，现在再加一台机子，负载就一定能到2倍么？其实时不一定的
4. 如果需要修改限流的值，虽然之前我们将令牌桶的限流时可以动态调整，但是靠人去调整，如果真出现问题然后再叫运维或者时开发同学去调整可能黄花菜都凉了

既然这种方式有这么多的缺点，那有没有办法解决呢？答案就是 自适应限流

2.自适应限流

2.1 自适应限流怎么做

前面我们遇到的主要问题就是每个服务实例的限流阈值实际应该是动态变化的，我们应该 根据系统能够承载的最大吞吐量，来进行限流，当当前的流量大于最大吞吐的时候就限制流量进入，反之则允许通过。

那现在的问题就是

• 系统的吞吐量该如何计算
• 什么时候系统的吞吐量就是最大的吞吐量了？

计算吞吐量：利特尔法则 L = λ * W

如上图所示，如果我们开一个小店，平均每分钟进店 2 个客人(λ)，每位客人从等待到完成交易需要 4 分钟(W)，那我们店里能承载的客人数量就是 2 * 4 = 8 个人

同理，我们可以将 λ 当做 QPS， W 呢是每个请求需要花费的时间，那我们的系统的吞吐就是 L = λ * W ，所以我们可以使用利特尔法则来计算系统的吞吐量。

什么时候系统的吞吐量就是最大的吞吐量？

首先我们可以通过统计过去一段时间的数据，获取到平均每秒的请求量，也就是 QPS，以及请求的耗时时间，为了避免出现前面 900ms 一个请求都没有最后 100ms 请求特别多的情况，我们可以使用滑动窗口算法来进行统计。

最容易想到的就是我们从系统启动开始，就把这些值给保存下来，然后计算一个吞吐的最大值，用这个来表示我们的最大吞吐量就可以了。但是这样存在一个问题是，我们很多系统其实都不是独占一台机器的，一个物理机上面往往有很多服务，并且一般还存在一些超卖，所以可能第一个小时最大处理能力是 100，但是这台节点上其他服务实例同时都在抢占资源的时候，这个处理能力最多就只能到 80 了.

所以我们需要一个数据来做启发阈值，只要这个指标达到了阈值那我们就进入流控当中。常见的选择一般是 CPU、Memory、System Load，这里我们以 CPU 为例

只要我们的 CPU 负载超过 80% 的时候，获取过去 5s 的最大吞吐数据，然后再统计当前系统中的请求数量，只要当前系统中的请求数大于最大吞吐那么我们就丢弃这个请求.

2.3 kratos 自适应限流分析

2.1 限流公式

// PS: 官方文档这里写的是 cpu > 800 AND (Now - PrevDrop) < 1s// 应该是写错了，等下看源码就知道了(cpu > 800 OR (Now - PrevDrop) < 1s) AND (MaxPass * MinRt * windows / 1000) < InFlight

• cup > 800

  表示CPU负载大于80% 进入限流

• (Now - PrevDrop) < 1s

  这个表示只要触发过 1 次限流，那么 1s 内都会去做限流的判定，这是为了避免反复出现限流恢复导致请求时间和系统负载产生大量毛刺

• (MaxPass * MinRt * windows / 1000) < InFlight

  判断当前负载是否大于最大负载

  InFlight

  表示当前系统中有多少请求

• (MaxPass * MinRt * windows / 1000)

  表示过去一段时间的最大负载

• MaxPass

  表示最近 5s 内，单个采样窗口中最大的请求数

• MinRt

  表示最近 5s 内，单个采样窗口中最小的响应时间

• windows

  表示一秒内采样窗口的数量，默认配置中是 5s 50 个采样，那么 windows 的值为 10。

2.2 源码分析

1. BBR结构体

   type BBR struct {cpu             cpuGetter// 请求数，和响应时间的采样数据，使用滑动窗口进行统计passStat        metric.RollingCounterrtStat          metric.RollingCounter// 当前系统中的请求数inFlight        int64// 每秒钟内的采样数量，默认是10winBucketPerSec int64// 单个 bucket 的时间bucketDuration  time.Duration// 窗口数量winSize         int// 配置conf            *ConfigprevDrop        atomic.Value// 表示最近 5s 内，单个采样窗口中最大的请求数的缓存数据maxPASSCache    atomic.Value// 表示最近 5s 内，单个采样窗口中最小的响应时间的缓存数据minRtCache      atomic.Value}

2. Allow: 判断请求是否允许通过

   func (l *BBR) Allow(ctx context.Context, opts ...limit.AllowOption) (func(info limit.DoneInfo), error) {// ... 省略配置修改代码if l.shouldDrop() {return nil, ecode.LimitExceed}atomic.AddInt64(&l.inFlight, 1)stime := time.Since(initTime)return func(do limit.DoneInfo) {rt := int64((time.Since(initTime) - stime) / time.Millisecond)l.rtStat.Add(rt)atomic.AddInt64(&l.inFlight, -1)switch do.Op {case limit.Success:l.passStat.Add(1)returndefault:return}}, nil}

   这个方法主要是给中间件使用的

   首先使用

   shouldDrop

   方法判断这个请求是否应该丢弃

3. 如果成功放行，那么当前系统中的请求数就 +1(原子操作)
4. 然后返回一个 function 用于请求结束之后的操作
   统计请求的响应时间
   • 如果统计成功了，给成功数请求数 +1
   • 并且当前系统种的请求数量

     Inflight

     -1

shouldDrop:

func (l *BBR) shouldDrop() bool {if l.cpu() < l.conf.CPUThreshold {prevDrop, _ := l.prevDrop.Load().(time.Duration)if prevDrop == 0 {return false}if time.Since(initTime)-prevDrop <= time.Second {inFlight := atomic.LoadInt64(&l.inFlight)return inFlight > 1 && inFlight > l.maxFlight()}l.prevDrop.Store(time.Duration(0))return false}inFlight := atomic.LoadInt64(&l.inFlight)drop := inFlight > 1 && inFlight > l.maxFlight()if drop {prevDrop, _ := l.prevDrop.Load().(time.Duration)if prevDrop != 0 {return drop}l.prevDrop.Store(time.Since(initTime))}return drop}

maxFlight

go func (l *BBR) maxFlight() int64 { return int64(math.Floor(float64(l.maxPASS()*l.minRT()*l.winBucketPerSec)/1000.0 + 0.5)) }