限流器系列(3)--自适应限流

2020-08-23
4分钟阅读时长

漏斗桶/令牌桶确实能够保护系统不被拖垮, 但不管漏斗桶还是令牌桶, 其防护思路都是设定一个指标, 当超过该指标后就阻止或减少流量的继续进入,当系统负载降低到某一水平后则恢复流量的进入。但其通常都是被动的,其实际效果取决于限流阈值设置是否合理,但往往设置合理不是一件容易的事情.

项目日常维护中, 经常能够看到某某同学在群里说:xx系统429了, 然后经过一番查找后发现是一波突然的活动流量, 只能申请再新增几台机器. 过了几天 OP 发现该集群的流量达不到预期又下掉了几台机器, 然后又开始一轮新的循环.

这里先不讨论集群自动伸缩的问题. 这里提出一些问题

  1. 集群增加机器或者减少机器限流阈值是否要重新设置?
  2. 设置限流阈值的依据是什么?
  3. 人力运维成本是否过高?
  4. 当调用方反馈429时, 这个时候重新设置限流, 其实流量高峰已经过了重新评估限流是否有意义?

这些其实都是采用漏斗桶/令牌桶的缺点, 总体来说就是太被动, 不能快速适应流量变化

自适应限流

对于自适应限流来说, 一般都是结合系统的 Load、CPU 使用率以及应用的入口 QPS、平均响应时间和并发量等几个维度的监控指标,通过自适应的流控策略, 让系统的入口流量和系统的负载达到一个平衡,让系统尽可能跑在最大吞吐量的同时保证系统整体的稳定性。

比较出名的自适应限流的实现是 Alibaba Sentinel. 不过由于提前没有发现 Sentinel 有个 golang 版本的实现, 本篇文章就以 Kratos 的 BBR 实现探讨自适应限流的原理.

Kratos 自适应限流

借鉴了 Sentinel 项目的自适应限流系统, 通过综合分析服务的 cpu 使用率、请求成功的 qps 和请求成功的 rt 来做自适应限流保护。

  • cpu: 最近 1s 的 CPU 使用率均值,使用滑动平均计算,采样周期是 250ms
  • inflight: 当前处理中正在处理的请求数量
  • pass: 请求处理成功的量
  • rt: 请求成功的响应耗时

限流公式

cpu > 800 AND (Now - PrevDrop) < 1s AND (MaxPass * MinRt * windows / 1000) < InFlight

  • MaxPass 表示最近 5s 内,单个采样窗口中最大的请求数
  • MinRt 表示最近 5s 内,单个采样窗口中最小的响应时间
  • windows 表示一秒内采样窗口的数量,默认配置中是 5s 50 个采样,那么 windows 的值为 10

kratos 中间件实现

func (b *RateLimiter) Limit() HandlerFunc {
	return func(c *Context) {
		uri := fmt.Sprintf("%s://%s%s", c.Request.URL.Scheme, c.Request.Host, c.Request.URL.Path)
		limiter := b.group.Get(uri)
		done, err := limiter.Allow(c)
		if err != nil {
			_metricServerBBR.Inc(uri, c.Request.Method)
			c.JSON(nil, err)
			c.Abort()
			return
		}
		defer func() {
			done(limit.DoneInfo{Op: limit.Success})
			b.printStats(uri, limiter)
		}()
		c.Next()
	}
}

使用方式

e := bm.DefaultServer(nil)
limiter := bm.NewRateLimiter(nil)
e.Use(limiter.Limit())
e.GET("/api", myHandler)

源码实现

Allow

func (l *BBR) Allow(ctx context.Context, opts ...limit.AllowOption) (func(info limit.DoneInfo), error) {
	allowOpts := limit.DefaultAllowOpts()
	for _, opt := range opts {
		opt.Apply(&allowOpts)
	}
	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) // 增加rtStat响应耗时的统计
		atomic.AddInt64(&l.inFlight, -1) // 请求处理成功后, 减少正在处理的请求数
		switch do.Op {
		case limit.Success:
			l.passStat.Add(1) // 处理成功后增加成功处理请求数的统计
			return
		default:
			return
		}
	}, nil
}

shouldDrop

func (l *BBR) shouldDrop() bool {
	// 判断目前cpu的使用率是否达到设置的CPU的限制, 默认值800
	if l.cpu() < l.conf.CPUThreshold { 
		// 如果上一次舍弃请求的时间是0, 那么说明没有限流的需求, 直接返回
		prevDrop, _ := l.prevDrop.Load().(time.Duration)
		if prevDrop == 0 {
			return false
		}
		// 如果上一次请求的时间与当前的请求时间小于1s, 那么说明有限流的需求
		if time.Since(initTime)-prevDrop <= time.Second {
			if atomic.LoadInt32(&l.prevDropHit) == 0 {
				atomic.StoreInt32(&l.prevDropHit, 1)
			}
			// 增加正在处理的请求的数量
			inFlight := atomic.LoadInt64(&l.inFlight)
			// 判断正在处理的请求数是否达到系统的最大的请求数量
			return inFlight > 1 && inFlight > l.maxFlight()
		}
		// 清空当前的prevDrop
		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

该函数是核心函数. 其计算公式: MaxPass * MinRt * windows / 1000. maxPASS/minRT都是基于metric.RollingCounter来实现的, 限于篇幅原因这里就不再具体看其实现(想看的可以去看rolling_counter_test.go还是蛮容易理解的)

func (l *BBR) maxFlight() int64 {
	return int64(math.Floor(float64(l.maxPASS()*l.minRT()*l.winBucketPerSec)/1000.0 + 0.5))
}
  • winBucketPerSec: 每秒内的采样数量,其计算方式:int64(time.Second)/(int64(conf.Window)/int64(conf.WinBucket)), conf.Window默认值10s, conf.WinBucket默认值100. 简化下公式: 1/(10/100) = 10, 所以每秒内的采样数就是10
// 单个采样窗口在一个采样周期中的最大的请求数, 默认的采样窗口是10s, 采样bucket数量100
func (l *BBR) maxPASS() int64 {
	rawMaxPass := atomic.LoadInt64(&l.rawMaxPASS)
	if rawMaxPass > 0 && l.passStat.Timespan() < 1 {
		return rawMaxPass
	}
	// 遍历100个采样bucket, 找到采样bucket中最大的请求数
	rawMaxPass = int64(l.passStat.Reduce(func(iterator metric.Iterator) float64 {
		var result = 1.0
		for i := 1; iterator.Next() && i < l.conf.WinBucket; i++ {
			bucket := iterator.Bucket()
			count := 0.0
			for _, p := range bucket.Points {
				count += p
			}
			result = math.Max(result, count)
		}
		return result
	}))
	if rawMaxPass == 0 {
		rawMaxPass = 1
	}
	atomic.StoreInt64(&l.rawMaxPASS, rawMaxPass)
	return rawMaxPass
}

// 单个采样窗口中最小的响应时间
func (l *BBR) minRT() int64 {
	rawMinRT := atomic.LoadInt64(&l.rawMinRt)
	if rawMinRT > 0 && l.rtStat.Timespan() < 1 {
		return rawMinRT
	}
	// 遍历100个采样bucket, 找到采样bucket中最小的响应时间
	rawMinRT = int64(math.Ceil(l.rtStat.Reduce(func(iterator metric.Iterator) float64 {
		var result = math.MaxFloat64
		for i := 1; iterator.Next() && i < l.conf.WinBucket; i++ {
			bucket := iterator.Bucket()
			if len(bucket.Points) == 0 {
				continue
			}
			total := 0.0
			for _, p := range bucket.Points {
				total += p
			}
			avg := total / float64(bucket.Count)
			result = math.Min(result, avg)
		}
		return result
	})))
	if rawMinRT <= 0 {
		rawMinRT = 1
	}
	atomic.StoreInt64(&l.rawMinRt, rawMinRT)
	return rawMinRT
}

参考

  1. alibaba/Sentinel-系统自适应限流
  2. go-kratos/kratos-自适应限流保护
  3. alibaba/sentinel-golang-系统自适应流控