Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

捉虫大师
• 阅读 1036

要说现在工程师最重要的能力,我觉得工程能力要排第一。

就算现在大厂面试经常要手撕算法,也是更偏向考查代码工程实现的能力,之前在群里看到这样的图片,就觉得很离谱(大概率是假的~)。

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

算法与工程实现

在 Sentinel-Go 中,一个很核心的算法是流控(限流)算法。

流控可能每个人都听过,但真要手写一个,还是有些困难。为什么流控算法难写?以我的感觉是算法和工程实现上存在一定差异,虽然算法好理解,但却没法照着实现。

举个例子,令牌桶算法很好理解,只需给定一个桶,以恒定的速率往桶内放令牌,满了则丢弃,执行任务前先去桶里拿令牌,只有拿到令牌才可以执行,否则拒绝。

如果实现令牌桶,按道理应该用一个单独线程(或进程)往桶里放令牌,业务线程去桶里取,但真要这么实现,怎么保证这个单独线程能稳定执行,万一挂了岂不是很危险?

所以工程实现上和算法原本肯定存在一定的差异,这也是为什么需要深入源码的一个原因。

滑动时间窗口的演进

通常来说,流控的度量是按每秒的请求数,也就是 QPS

QPS:query per second,指每秒查询数,当然他的意义已经泛化了,不再特指查询,可以泛指所有请求。如果非要区分,TPS 指每秒事务数,即写入数,或 RPS,每秒请求数,本文不分这么细,统一叫QPS。

当然也有按并发数来度量,并发数的流控就非常简单

并发数流控

并发是一个瞬时概念,它跟时间没有关系。和进程中的线程数、协程数一样,每次取的时候只能拿到一个瞬间的快照,但可能很快就变化了。

并发数怎么定义?可以近似认为进入业务代码开始就算一个并发,执行完这个并发就消失。

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

这样说来,实现就非常简单了,只需要定义一个全局变量,责任链开始时对这个变量原子增1,并获取当前并发数的一个快照,判断并发数是否超限,如果超限则直接阻断,执行完了别忘了原子减1即可,由于太过简单,就不需要放代码了。

固定时间窗口

参考并发数流控,当需要度量 QPS 时,是否也可以利用这样的思想呢?

由于 QPS 有时间的度量,第一直觉是和并发数一样弄个变量,再起个单独线程每隔 1s 重置这个变量。

但单独线程始终不放心,需要稍微改一下。

如果系统有一个起始时间,每次请求时,获取当前时间,两者之差,就能算出当前处于哪个时间窗口,这个时间窗口单独计数即可。

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

如果稍微思考下,你会发现问题不简单,如下图,10t 到20t 只有60个请求,20t到30t之间只有80个请求,但有可能16t到26t之间有110个请求,这就很有可能把系统打垮。

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

滑动时间窗口

为了解决上面的问题,工程师想出了一个好办法:别固定时间窗口,以当前时间往前推算窗口

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

但问题又来了,这该怎么实现呢?

滑动时间窗口工程实现

在工程实现上,可以将时间划分为细小的采样窗口,缓存一段时间的采样窗口,这样每当请求来的时候,只需要往前拿一段时间的采样窗口,然后求和就能拿到总的请求数。

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

Sentinel-Go 滑动时间窗口的实现

前方代码高能预警~

Sentinel-Go 是基于 LeapArray 实现的滑动窗口,其数据结构如下

type LeapArray struct {
    bucketLengthInMs uint32 // bucket大小
    sampleCount      uint32 // bucket数量
    intervalInMs     uint32 // 窗口总大小
    array            *AtomicBucketWrapArray // bucket数组
    updateLock mutex // 更新锁
}

type AtomicBucketWrapArray struct {
    base unsafe.Pointer // 数组的起始地址
    length int // 长度,不能改变
    data   []*BucketWrap // 真正bucket的数据
}

type BucketWrap struct {
    BucketStart uint64 // bucket起始时间
    Value atomic.Value // bucket数据结构,例如 MetricBucket
}

type MetricBucket struct {
    counter        [base.MetricEventTotal]int64 // 计数数组,可放不同类型
    minRt          int64 // 最小RT
    maxConcurrency int32 // 最大并发数
}

再看下是如何写入指标的,例如当流程正常通过时

// ①
sn.AddCount(base.MetricEventPass, int64(count))

// ②
func (bla *BucketLeapArray) AddCount(event base.MetricEvent, count int64) {
    bla.addCountWithTime(util.CurrentTimeMillis(), event, count)
}

// ③
func (bla *BucketLeapArray) addCountWithTime(now uint64, event base.MetricEvent, count int64) {
    b := bla.currentBucketWithTime(now)
    if b == nil {
        return
    }
    b.Add(event, count)
}

// ④
func (mb *MetricBucket) Add(event base.MetricEvent, count int64) {
    if event >= base.MetricEventTotal || event < 0 {
        logging.Error(errors.Errorf("Unknown metric event: %v", event), "")
        return
    }
    if event == base.MetricEventRt {
        mb.AddRt(count)
        return
    }
    mb.addCount(event, count)
}

// ⑤
func (mb *MetricBucket) addCount(event base.MetricEvent, count int64) {
    atomic.AddInt64(&mb.counter[event], count)
}

取到相应的 bucket,然后写入相应 event 的 count,对 RT 会特殊处理,因为有一个最小 RT 需要处理。

重点看是如何取到相应的 bucket 的:

func (bla *BucketLeapArray) currentBucketWithTime(now uint64) *MetricBucket {
    // ①根据当前时间取bucket
    curBucket, err := bla.data.currentBucketOfTime(now, bla)
    ...
    b, ok := mb.(*MetricBucket)
    if !ok {
        ...
        return nil
    }
    return b
}

func (la *LeapArray) currentBucketOfTime(now uint64, bg BucketGenerator) (*BucketWrap, error) {
    ...
    // ②计算index = (now / bucketLengthInMs) % LeapArray.array.length
    idx := la.calculateTimeIdx(now)
    // ③计算bucket开始时间 = now - (now % bucketLengthInMs)
    bucketStart := calculateStartTime(now, la.bucketLengthInMs)

    for { 
        old := la.array.get(idx)
        if old == nil { // ④未使用,直接返回
            newWrap := &BucketWrap{
                BucketStart: bucketStart,
                Value:       atomic.Value{},
            }
            newWrap.Value.Store(bg.NewEmptyBucket())
            if la.array.compareAndSet(idx, nil, newWrap) {
                return newWrap, nil
            } else {
                runtime.Gosched()
            }
        } else if bucketStart == atomic.LoadUint64(&old.BucketStart) { // ⑤刚好取到是当前bucket,返回
            return old, nil
        } else if bucketStart > atomic.LoadUint64(&old.BucketStart) { // ⑥取到了旧的bucket,重置使用
            if la.updateLock.TryLock() {
                old = bg.ResetBucketTo(old, bucketStart)
                la.updateLock.Unlock()
                return old, nil
            } else {
                runtime.Gosched()
            }
        } else if bucketStart < atomic.LoadUint64(&old.BucketStart) { // ⑦取到了比当前还新的bucket,总共只有一个bucket时,并发情况可能会出现这种情况,其他情况不可能,直接报错
            if la.sampleCount == 1 {
                return old, nil
            }

            return nil, errors.New(fmt.Sprintf("Provided time timeMillis=%d is already behind old.BucketStart=%d.", bucketStart, old.BucketStart))
        }
    }
}

举个直观的例子,看如何拿到 bucket:

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

  • 假设 B2 取出来是 nil,则 new 一个 bucket 通过 compareAndSet 写入,保证线程安全,如果别别的线程先写入,这里会执行失败,调用 runtime.Gosched(),让出时间片,进入下一次循环
  • 假设取出 B2 的开始时间是3400,与计算的相同,则直接使用
  • 假设取出的 B2 的开始时间小于 3400,说明这个 bucket 太旧了,需要覆盖,使用更新锁来更新,保证线程安全,如果拿不到锁,也让出时间片,进入下一次循环
  • 假设取出 B2 的开始时间大于3400,说明已经有其他线程更新了,而 bucketLengthInMs 通常远远大于锁的获取时间,所以这里只考虑只有一个 bucket 的情况直接返回,其他情况报错

回到 QPS 计算:

qps := stat.InboundNode().GetQPS(base.MetricEventPass)

该方法会先计算一个起始时间范围

func (m *SlidingWindowMetric) getBucketStartRange(timeMs uint64) (start, end uint64) {
    curBucketStartTime := calculateStartTime(timeMs, m.real.BucketLengthInMs())
    end = curBucketStartTime
    start = end - uint64(m.intervalInMs) + uint64(m.real.BucketLengthInMs())
    return
}

例如当前时间为3500,则计算出

  • end = 3400
  • start = 3400 - 1200 + 200 = 2400

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

然后遍历所有 bucket,把在这个范围内的 bucket 都拿出来,计算 QPS,只需要相加即可。

最后

本节从滑动窗口流控算法的工程实现演进到 Sentinel-Go 里滑动窗口的实现,从 Sentinel-Go 的实现上看到,还得考虑内存的使用,并发控制等等,如果完全写出来,还是非常不容易的。

《Sentinel-Go源码系列》已经写了三篇,只介绍了两个知识点:责任链模式、滑动窗口限流,后续还有对象池等,但这其实和 Sentinel-Go 关系不是很大,到时候单独成文,就不放在本系列里了。

本文算是一个结束,与其说是结束,不如说是一个开始。


  • 搜索关注微信公众号"捉虫大师",后端技术分享,架构设计、性能优化、源码阅读、问题排查、踩坑实践。
  • 另外我也准备组建一个技术交流群,但现在也不知道会有多少人,所以大家先加我微信 MrRoshi,备注加群,一起交流技术,超过一定人数我就拉一个~

Sentinel-Go 源码系列(三)滑动时间窗口算法的工程实现

点赞
收藏
评论区
推荐文章
胡哥有话说 胡哥有话说
2年前
前端面试题解密:经典算法之冒泡算法(ES6版)及优化
前言随着前端的飞速发展,前端业务开发给前端工程师提出了更高的要求,因而算法题也越来越高频次的出现在前端面试中。有很多的小伙伴找胡哥苦诉,在前端实际开发中(除了涉及游戏开发方面),算法使用有很多吗?大厂的面试是故意要自我标榜下吗?其实不然,考核算法还是相当有必要的,来来来,让胡哥给你拯救世界的理由,哦,不,是考核算法的理由。为啥要考算法?1.算法是
鲍忠 鲍忠
1年前
网易微专业高级前端开发工程师2022克鲁斯卡尔算法和普里姆算法
网易微专业高级前端开发工程师2022克鲁斯卡尔算法和普里姆算法createvue的实现原理源代码既然命令vue@x是通过npminit执行的,那么npminit做了什么?它怎么知道有vue@x这个命令呢?这是一个性的问题。先解决
Stella981 Stella981
2年前
Github惊现标星68K的力扣算法刷题宝典,再也不怕被大厂算法拦路了
写在前面BAT等国内的一线名企,在招聘工程师的过程中,对算法和数据结构都会重点考察。但算法易学难精,我的很多粉丝技术能力不错,但面试时总败在算法这一关,拿不到好Offer。但说实话,数据结构和算法花点时间,用对方法,很容易解决。面试官为什么爱问数据结构与算法,答案很简单:算法能力能够准确辨别一个程序员的技术功底是
Stella981 Stella981
2年前
Sentinel 是如何做限流的
限流是保障服务高可用的方式之一,尤其是在微服务架构中,对接口或资源进行限流可以有效地保障服务的可用性和稳定性。之前的项目中使用的限流措施主要是Guava的RateLimiter。RateLimiter是基于令牌桶流控算法,使用非常简单,但是功能相对比较少。而现在,我们有了一种新的选择,阿里提供的Sentinel。Sentinel是阿里巴巴提供
Stella981 Stella981
2年前
Spring Boot的接口限流应用
阅读目录:1\.前言2\.算法介绍计数器法3\.算法介绍滑动窗口4\.算法介绍漏桶算法5\.算法介绍令牌桶算法前言在一个高并发系统中对流量的把控是非常重要的,当巨大的流量直接请求到我们的服务器上没多久就可能造成接口不可用
稚然 稚然
3星期前
【开课吧】算法工程师-高级深度学习
//下仔のke:https://yeziit.cn/13880/算法工程师是指专门从事算法研究和开发的专业人员。他们主要负责设计和实现算法,解决实际应用中的问题,如机器学习、数据挖掘、图像处理等。算法工程师需要具备扎实的数学基础、编程能力、算法设计和分析能
常用限流算法详解
一、有哪些常用的限流算法1.固定窗口限流;2.滑动窗口限流;3.漏桶算法限流;4.令牌桶算法限流。二、4种限流算法介绍1.固定窗口限流举例说明:假设时间窗口大小为5s,则0到5s为第一个窗口,5到10s为第二个窗
贾蓁 贾蓁
1个月前
开课吧-算法工程师-高级深度学习
开课吧算法工程师高级深度学习算法工程师的介绍download》http://quangneng.com/2405/算法工程师是一种专注于设计、实现和优化算法的专业人士。他们在计算机科学领域发挥关键作用,致力于解决各种复杂问题,从而提高计算效率、性能和准确性
稚然 稚然
3星期前
比特鹏哥高阶算法班2023
//下仔のke:https://yeziit.cn/15008/嵌入式工程师在工作中涉及的算法主要与嵌入式系统的性能优化和资源管理相关。例如,为了确保系统的实时性、稳定性、能耗等方面达到最优,嵌入式工程师需要设计和实现各种算法。这些算法可能涉及到时间序列分
笑面虎 笑面虎
3星期前
LLM大语言模型算法特训,带你转型AI大语言模型算法工程师
LLM大语言模型算法特训,带你转型AI大语言模型算法工程师给大家分享一套课程——LLM大语言模型算法特训,带你转型AI大语言模型算法工程师,源码PDF课件下载。大语言模型(LLM)是基于海量文本数据训练的深度学习模型。它不仅能够生成自然语言文本,还能够深