如何在 Golang 自适应微服务自助服务背后实现滑动窗口算法

terry 2年前 (2023-09-24) 阅读数 105 #后端开发

了解如何实现 go-zero 框架封装的滑动窗口算法。 RollingWindow 是一种数据结构，用于计算某个时间间隔内事件的段。该滑动窗口使用循环缓冲区来存储桶，每个桶代表一个间隔内的一个时间。例如：最近30秒成功请求数、请求总数。

适用于需要依赖某些数字指示器的场景，例如自适应电流中断和保险丝。

01 基本思想

将整个时间间隔totalInterval分为n段，每段时间间隔为一个区间，各段存储在长度为n的循环数组中。假设 totalInterval = 10s，n = 40，则间隔 = totalInterval / n = 250 ms。每次添加新的段时，都需要计算自上次更新以来已经切换了多少个段，更新环形缓冲区偏移量，并重置过期的段。？在中，所需段的数量由开发人员定义。 go-zero框架fuse实现googlebreaker默认使用40个桶。每个段中定义了两个参数：Sum 代表成功总数，Count 代表请求总数。对于断路器中的计算，将累计和Sum记录为Accepted，将累计和Count记录为total，从而计算出当前的错误率。？提供了Segment Set和Reduce方法来对所有Segment执行特定操作。

04 RollingWindow实现

我们来看看RollingWindow提供了哪些方法：

// NewRollingWindow 返回一个 RollingWindow，它有 size 个桶和时间间隔，使用 opts 来定制 RollingWindow。func NewRollingWindow(size int, interval time.Duration, opts ...RollingWindowOption) *RollingWindow {  if size < 1 {    panic("size must be greater than 0")  }
  w := &RollingWindow{    size:          size,    win:           newWindow(size),    interval:      interval,    offset:        0,    ignoreCurrent: false,    lastTime:      timex.Now(),  }  for _, opt := range opts {    opt(w)  }  return w}
// Add 添加值到当前桶// 通过 lastTime 和 nowTime 的标注，不断重置来实现窗口滑动，新的数据不断补上，从而实现滑动窗口的效果。func (rw *RollingWindow) Add(v float64) {  rw.lock.Lock()  defer rw.lock.Unlock()  // 滑动在此处进行  rw.updateOffset()  rw.win.add(rw.offset, v)}
// Reduce 运行 fn 在所有的桶上，如果 ignoreCurrent 被设置，则忽略当前桶。func (rw *RollingWindow) Reduce(fn func(b *Bucket)) {  rw.lock.RLock()  defer rw.lock.RUnlock()
  var diff int  span := rw.span()  // 忽略当前桶，因为数据不完整  if span == 0 && rw.ignoreCurrent {    diff = rw.size - 1  } else {    diff = rw.size - span  // size - span 取出未过期的桶  }  if diff > 0 {    offset := (rw.offset + span + 1) % rw.size    // 从 offset 开始，取出 diff 个桶    rw.win.reduce(offset, diff, fn)  }}
// span 自上次更新 lastTime 跨越了几个 bucketfunc (rw *RollingWindow) span() int {  offset := int(timex.Since(rw.lastTime) / rw.interval)  if 0 <= offset && offset < rw.size {    return offset  }
  return rw.size}
// updateOffset 更新环形缓冲区的偏移量。func (rw *RollingWindow) updateOffset() {  span := rw.span()  if span <= 0 {    return  }
  offset := rw.offset  // 重置过期的桶  for i := 0; i < span; i++ {    rw.win.resetBucket((offset + i + 1) % rw.size)  }
  rw.offset = (offset + span) % rw.size  now := timex.Now()  // 更新时间，与时间间隔对齐  rw.lastTime = now - (now-rw.lastTime)%rw.interval}

NewRollingWindow：用于创建滚动窗口对象，可以使用size（桶数）和interval（时间间隔）参数进行自定义设置。
添加：用于向当前段添加值。通过lastTime和nowTime注解，不断刷新窗口实现滑动窗口，不断添加新数据实现滑动窗口效果。
Reduce：用于在所有桶上运行指定的函数。如果设置了ignoreCurrent选项，则当前段将被忽略。

在断路器中，Reduce 函数用于计算接受值和输入值。

func (b *googleBreaker) history() (accepts, total int64) {  b.stat.Reduce(func(b *collection.Bucket) {    accepts += int64(b.Sum)    total += b.Count  })
  return}

updateOffset：这是一个比较基础的方法。这是滚动完成的地方。它用于更新段、重置过期段以及确定当前时间必须落入哪个段。
- 确定当前时间相对于分段间隔的范围，即调用range方法。
- 清除范围内所有段的数据，因为它已经过期了。
- 更新offset的当前偏移量，这也是数据将要写入的段。
- 更新最后一项并标记下一步。