go-sentinel流量控制(二)：sentinel统计数据结构揭秘

介绍

通常情况下，一个流量治理组件具备基本的限流能力，其核心思想是统计一定时间内的请求数，然后根据预设的阈值来判断是否需要进行限流操作。

在实现这一思想时，如何有效地存储和统计一段时间内的请求数是至关重要的。本文将重点介绍Sentinel-Go是如何实现指标存储和数据统计的。

固定窗口算法

固定窗口算法是实现流量控制的一种简单方式，也称为计数器算法。通过原子操作在统计周期内累加请求次数，当请求数超过阈值时进行限流操作。

实现代码：

var (
    counter    int64 //计数
    intervalMs int64 = 1000 //窗口长度(1S)
    threshold  int64 = 2 //限流阈值
    startTime        = time.Now().UnixMilli() //窗口开始时间
)

func main() {
    for i := 0; i < 10; i++ {
       if tryAcquire() {
          fmt.Println("成功请求", time.Now().Unix())
      }
   }
}

func tryAcquire() bool {
    if time.Now().UnixMilli()-atomic.LoadInt64(&startTime) > intervalMs {
       atomic.StoreInt64(&startTime, time.Now().UnixMilli())
       atomic.StoreInt64(&counter, 0)
   }
    return atomic.AddInt64(&counter, 1) <= threshold
}

在固定窗口算法中存在边界问题，例如在一个1秒的统计周期内，限流阈值为2，如果4次请求正好跨越了固定的时间窗口，则会出现不符合限流预期的情况。

滑动时间窗口

滑动时间窗口算法则可以解决固定窗口算法的边界问题。滑动时间窗口算法中的重要概念包括统计周期和窗口大小。通过划分不同大小的窗口，在时间上进行滑动统计，可以提高统计的精确度。

在滑动窗口算法中，根据业务情况设定统计周期和窗口大小。窗口的大小影响统计的精确度和并发性能，窗口越大并发性能越好但精确度较低，窗口越小精确度较高但并发性能会降低。不同的统计周期和窗口大小组合可以应对不同的业务场景和流量特征。

Sentinel-Go 统计结构

下面将详细介绍 Sentinel-Go 是如何使用滑动时间窗口高效的存储和统计指标数据的

窗口结构

在Sentinel-Go中，滑动时间窗口实现了高效的存储和统计指标数据。窗口结构（BucketWrap）是滑动时间窗口的核心组成部分，包含了开始时间和抽象的统计结构。

type BucketWrap struct {
   // BucketStart represents start timestamp of this statistic bucket wrapper.
   BucketStart uint64
   // Value represents the actual data structure of the metrics (e.g. MetricBucket).
   Value atomic.Value
}

每个BucketWrap包括以下信息:

• BucketStart：表示当前统计窗口的起始时间
• Value：存储实际的指标数据结构，采用原子操作确保并发安全

通过这样的设计，Sentinel-Go可以有效地实现滑动时间窗口的统计功能。在该设计中，假设统计周期为1秒，每个窗口的长度为200毫秒。这样的设计能够提高统计的准确性和并发安全性，使得Sentinel-Go在处理流量控制时表现出色。

指标数据:

1. pass: 表示到来的数量，即此刻通过 Sentinel-Go 规则的流量数量
2. block: 表示被拦截的流量数量
3. complete: 表示完成的流量数量，包含正常结束和异常结束的情况
4. error: 表示错误的流量数量（熔断场景使用）
5. rt： 单次请求的request time
6. total：暂时无用

原子时间轮

为了解决滑动时间窗口统计中时间轮持续扩张的问题，Sentinel Go引入了原子时间轮的概念。原子时间轮通过一种巧妙的设计来优化时间窗口的统计过程，避免不断扩张的slice容量问题。

原子时间轮的设计包括以下要点：

• 使用固定大小的slice来表示整个统计周期内的时间窗口，每个元素对应一个窗口。
• 通过原子操作实现时间轮的滑动：将时间轮中的窗口向右移动一个单位，并将新的窗口数据初始化。
• 周期性地重置时间轮：当时间周期结束时，重置时间轮，重新开始新的统计周期。

通过引入原子时间轮的设计，Sentinel Go能够高效地实现时间窗口的滑动统计，避免slice容量无限增长的问题，同时确保统计数据的准确性和并发安全性。这种工程技术的创新为Sentinel Go在流量控制和限流方面提供了更好的性能和可靠性。

如下：原子时间轮数据结构

type AtomicBucketWrapArray struct {
   // The base address for real data array
   base unsafe.Pointer // 窗口数组首元素地址
   // The length of slice(array), it can not be modified.
   length int // 窗口数组的长度
   data   []*BucketWrap //窗口数组
}

初始化

1. 根据当前时间计算出当前时间对应的窗口的startime，并得到当前窗口对应的位置：

// 计算开始时间
func calculateStartTime(now uint64, bucketLengthInMs uint32) uint64 {
   return now - (now % uint64(bucketLengthInMs))
}
// 窗口下标位置
idx := int((now / uint64(bucketLengthInMs)) % uint64(len))

1. 初始化窗口数据结构（BucketWrap）

for i := idx; i <= len-1; i++ {
   ww := &BucketWrap{
      BucketStart: startTime,
      Value:       atomic.Value{},
   }
   ww.Value.Store(generator.NewEmptyBucket())
   ret.data[i] = ww
   startTime += uint64(bucketLengthInMs)
}
for i := 0; i < idx; i++ {
   ww := &BucketWrap{
      BucketStart: startTime,
      Value:       atomic.Value{},
   }
   ww.Value.Store(generator.NewEmptyBucket())
   ret.data[i] = ww
   startTime += uint64(bucketLengthInMs)
}

3. 将窗口数组首元素地址设置到原子时间轮：

// calculate base address for real data array
sliHeader := (*util.SliceHeader)(unsafe.Pointer(&ret.data))
ret.base = unsafe.Pointer((**BucketWrap)(unsafe.Pointer(sliHeader.Data)))

这段代码的目的是将窗口数组（slice）的第一个元素（即第一个时间窗口）的地址设置到原子时间轮中，以便实现对时间轮中的元素进行原子无锁的读取和更新操作。

通过使用unsafe.Pointer进行地址操作，可以将底层slice的首元素地址转换为**BucketWrap类型，然后将该地址设置到原子时间轮中。这样做的好处在于可以通过原子操作直接访问和更新时间轮中的元素，而无需使用锁来保证并发安全性。

这种技术手段虽然使用了unsafe包中的功能，并且需要对指针操作有一定的了解，但在需要高效并发处理的场景下，能够提高代码的执行效率和性能。然而，使用unsafe包需要谨慎，确保操作的安全性和正确性。

窗口获取&窗口替换

如何在并发安全的情况下读取窗口和对窗口进行替换（时间轮涉及到对窗口更新操作），代码如下：

// 获取对应窗口
func (aa *AtomicBucketWrapArray) get(idx int) *BucketWrap {
   // aa.elementOffset(idx) return the secondary pointer of BucketWrap, which is the pointer to the aa.data[idx]
   // then convert to (*unsafe.Pointer)
   if offset, ok := aa.elementOffset(idx); ok {
      return (*BucketWrap)(atomic.LoadPointer((*unsafe.Pointer)(offset)))
   }
   return nil
}

// 替换对应窗口
func (aa *AtomicBucketWrapArray) compareAndSet(idx int, except, update *BucketWrap) bool {
   // aa.elementOffset(idx) return the secondary pointer of BucketWrap, which is the pointer to the aa.data[idx]
   // then convert to (*unsafe.Pointer)
   // update secondary pointer
   if offset, ok := aa.elementOffset(idx); ok {
      return atomic.CompareAndSwapPointer((*unsafe.Pointer)(offset), unsafe.Pointer(except), unsafe.Pointer(update))
   }
   return false
}

// 获取对应窗口的地址
func (aa *AtomicBucketWrapArray) elementOffset(idx int) (unsafe.Pointer, bool) {
   if idx >= aa.length || idx < 0 {
      logging.Error(errors.New("array index out of bounds"),
         "array index out of bounds in AtomicBucketWrapArray.elementOffset()",
         "idx", idx, "arrayLength", aa.length)
      return nil, false
   }
   basePtr := aa.base
   return unsafe.Pointer(uintptr(basePtr) + uintptr(idx)*unsafe.Sizeof(basePtr)), true
}

获取窗口

根据描述，获取窗口的过程可以分为以下步骤：

1. 首先，根据当前时间计算出窗口对应的下标位置。
2. 在获取函数（get func）中调用elementOffset函数，该函数的作用是根据下标位置定位相应窗口的元素地址。
3. 在elementOffset函数中，首先将底层slice的首元素地址转换为uintptr类型。
4. 然后，根据窗口对应下标乘以每个元素的指针字节大小，可以得到相应窗口元素的地址。
5. 将获取的窗口元素地址转换为时间窗口（*BucketWrap类型），即可获得对应时间窗口的数据。

这个过程主要通过底层指针操作和地址计算来实现对窗口元素的定位，确保能够准确地获取指定时间窗口的数据。这种设计可以在保证并发安全性的前提下高效地访问和更新时间窗口数据，适用于需要高性能并发处理的场景，比如流量控制、统计和限流等应用中。

窗口更新：和获取窗口一样，获取到对应下标位置的窗口地址，然后利用 atomic.CompareAndSwapPointer 进行 CAS 更新，将新的窗口指针地址更新到底层数组中。

滑动窗口

在原子时间轮中提供了对窗口读取以及更新的操作。那么在什么时机触发更新以及如何滑动呢？

滑动

所谓滑动就是根据当前时间找到整个统计周期的所有窗口中的数据。例如在限流场景下，我们需要获取统计周期内的所有pass的流量，从而来判断当前流量是否应该被限流。

核心代码如下：

// 根据当前时间获取周期内的所有窗口
func (m *SlidingWindowMetric) getSatisfiedBuckets(now uint64) []*BucketWrap {
   start, end := m.getBucketStartRange(now)
   satisfiedBuckets := m.real.ValuesConditional(now, func(ws uint64) bool {
      return ws >= start && ws <= end
   })
   return satisfiedBuckets
}


// 根据当前时间获取整个周期对应的窗口的开始时间和结束时间
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
}

// 匹配符合条件的窗口
func (la *LeapArray) ValuesConditional(now uint64, predicate base.TimePredicate) []*BucketWrap {
   if now <= 0 {
      return make([]*BucketWrap, 0)
   }
   ret := make([]*BucketWrap, 0, la.array.length)
   for i := 0; i < la.array.length; i++ {
      ww := la.array.get(i)
      if ww == nil || la.isBucketDeprecated(now, ww) || !predicate(atomic.LoadUint64(&ww.BucketStart)) {
         continue
      }
      ret = append(ret, ww)
   }
   return ret
}

如下图所示：统计周期=1000ms（跨两个格子），now=1300时 计算出 start=500,end=1000:

更新

每次流量经过时都会进行相应的指标存储，在存储时会先获取对应的窗口，然后会根据窗口的开始时间进行对比，如果过期则进行窗口重置。

如下图：根据窗口开始时间匹配发现0号窗口已过期：

如下图：重置窗口的开始时间和统计指标：

核心代码：

func (la *LeapArray) currentBucketOfTime(now uint64, bg BucketGenerator) (*BucketWrap, error) {
   // 计算当前时间对应的窗口下标
   idx := la.calculateTimeIdx(now)
   // 计算当前时间对应的窗口的开始时间
   bucketStart := calculateStartTime(now, la.bucketLengthInMs)

   for {
     // 获取旧窗口
      old := la.array.get(idx)
      // 如果旧窗口==nil则初始化(正常不会执行这部分代码)
      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) {
         return old, nil
      //  如果本次计算的开始时间大于旧窗口的开始时间，则认为窗口过期尝试重置
      } else if bucketStart > atomic.LoadUint64(&old.BucketStart) {
         if la.updateLock.TryLock() {
            old = bg.ResetBucketTo(old, bucketStart)
            la.updateLock.Unlock()
            return old, nil
         } else {
            runtime.Gosched()
         }
        ......
      } 
}