熔断器
go-zero 的熔断器基于滑动窗口来实现，我们首先来看看滑动窗口是啥
leetcode 中有这样一个题目：
给定一个整数数据流和一个窗口大小，根据该滑动窗口的大小，计算滑动窗口里所有数字的平均值。
实现 MovingAverage 类：
MovingAverage(int size) 用窗口大小 size 初始化对象。
double next(int val) 成员函数 next 每次调用的时候都会往滑动窗口增加一个整数，请计算并返回数据流中最后 size 个值的移动平均值，即滑动窗口里所有数字的平均值。
示例：
输入：
inputs = ["MovingAverage", "next", "next", "next", "next"]
inputs = [[3], [1], [10], [3], [5]]
输出：
[null, 1.0, 5.5, 4.66667, 6.0]
解释：
MovingAverage movingAverage = new MovingAverage(3);
movingAverage.next(1); // 返回 1.0 = 1 / 1
movingAverage.next(10); // 返回 5.5 = (1 + 10) / 2
movingAverage.next(3); // 返回 4.66667 = (1 + 10 + 3) / 3
movingAverage.next(5); // 返回 6.0 = (10 + 3 + 5) / 3
我们来想一想解题思路：
- 窗口大小是固定的
- 窗口每次都会滑动
- 窗口滑动是替换就数据
我们来解一解题：
type MovingAverage struct {
index int // 当前环形数组的位置
count int // 数组大小
sum int // 数据总量
buckets []int // 环形数组
}
/** Initialize your data structure here. */
func Constructor(size int) MovingAverage {
return MovingAverage{index: size - 1, buckets: make([]int, size)}
}
func (ma *MovingAverage) Next(val int) float64 {
ma.sum += val
ma.index = (ma.index + 1) % len(ma.buckets) // 循环数组索引
if ma.count < len(ma.buckets) {
ma.count++
ma.buckets[ma.index] = val
} else {
ma.sum -= ma.buckets[ma.index] // 减去旧数据
ma.buckets[ma.index] = val // 替换旧数据
}
return float64(ma.sum) / float64(ma.count)
}
func Test_Demo(t *testing.T) {
ma := Constructor(3)
fmt.Println(ma.Next(1)) // 返回 1.0 = 1 / 1
fmt.Println(ma.Next(10)) // 返回 5.5 = (1 + 10) / 2
fmt.Println(ma.Next(3)) // 返回 4.66667 = (1 + 10 + 3) / 3
fmt.Println(ma.Next(5)) // 返回 6.0 = (10 + 3 + 5) / 3
}
从解题的代码中我们可以看到滑动窗口的本质是循环数组，而循环数组的核心思路是
- 循环数组的索引
ma.index = (ma.index + 1) % len(ma.cache) // 循环数组索引
- 新数据替换旧数据
ma.sum -= ma.cache[ma.index] // 减去旧数据
ma.cache[ma.index] = val // 替换旧数据
再来看看 go-zero 的 rollingwidnow，是不是和前面学习的滑动窗口是一样一样的呀 : )
type window struct {
buckets []*Bucket // 环形数组
size int
}
// 初始化窗口
func newWindow(size int) *window {
buckets := make([]*Bucket, size)
for i := 0; i < size; i++ {
buckets[i] = new(Bucket)
}
return &window{
buckets: buckets,
size: size,
}
}
// 往执行的 bucket 加入指定的指标数据
func (w *window) add(offset int, v float64) {
// 窗口滑动代码
// rw.offset = (offset + span) % rw.size
w.buckets[offset%w.size].add(v)
}
滑动窗口看完了，我们再来看看柳暗花明又一村的
其算法数学表达式如下：
- requests：请求数量（调用方发起请求的数量总和）
- accepts：请求接受数量（被调用方正常处理的请求数量）
- K：倍值（越小越敏感）
// 判断是否触发熔断
func (b *googleBreaker) accept() error {
accepts, total := b.History()
weightedAccepts := b.k * float64(accepts)
// Google Sre过载保护算法 https://landing.google.com/sre/sre-book/chapters/handling-overload/#eq2101
dropRatio := math.Max(0, (float64(total-protection)-weightedAccepts)/float64(total+1))
if dropRatio <= 0 {
return nil
}
if b.proba.TrueOnProba(dropRatio) {
return ErrServiceUnavailable
}
return nil
}
go-zero 熔断器给我们提供如下方法，更我们使用：
type (
// 自定义判定执行结果
Acceptable func(err error) bool
// 手动回调
Promise interface {
// Accept tells the Breaker that the call is successful.
// 请求成功
Accept()
// Reject tells the Breaker that the call is failed.
// 请求失败
Reject(reason string)
}
Breaker interface {
// 熔断器名称
Name() string
// 熔断方法，执行请求时必须手动上报执行结果
// 适用于简单无需自定义快速失败，无需自定义判定请求结果的场景
// 相当于手动挡。。。
Allow() (Promise, error)
// 熔断方法，自动上报执行结果
// 自动挡。。。
Do(req func() error) error
// 熔断方法
// acceptable - 支持自定义判定执行结果
DoWithAcceptable(req func() error, acceptable Acceptable) error
// 熔断方法
// fallback - 支持自定义快速失败
DoWithFallback(req func() error, fallback func(err error) error) error
// 熔断方法
// fallback - 支持自定义快速失败
// acceptable - 支持自定义判定执行结果
DoWithFallbackAcceptable(req func() error, fallback func(err error) error, acceptable Acceptable) error
}
)
关于 go-zero 熔断器的文章就到这里啦，看完之后是不是觉得很简单，觉得不简单可以多读几遍，感谢大家的阅读。
引用文章：