做为本文的前言，首先向读者介绍一下降级、熔断和限流的概念与关系。也许很多人对此，早已谙熟于心，但是烦请允许我再啰嗦几句，方便第一次接触该领域的小伙伴们，都可以的理解消化本文。
所谓限流，本质就是对系统的被请求频率以及内部的部分功能的执行频率加以限制，防止因突发的流量激增，导致整个系统不可用。当流量出现激增，触发限流，那么对于那些系统暂时不想或无法处理的“流量”，我们该如何处理呢？这就自然引出了服务降级的概念，其本质就是提供降低系统正常运行所能提供的功能数，亦或是降低某些功能完成的完整度（质量）。而熔断就是众多降级手段中最常见的一种，其在流量过大时（或下游服务出现问题时），可以自动断开与下游服务的交互，并可以通过自我诊断下游系统的错误是否已经修正，或上游流量是否减少至正常水平，来恢复自我恢复。
简而言之，限流是从系统的流量入口考虑，从进入的流量上进行限制，达到保护系统的作用；降级，是从系统内部的平级服务或者业务的维度考虑，流量大了，可以干掉一些，保护其他正常使用；熔断强调的是服务之间的调用能实现自我恢复的状态；
计数器是一种最简单限流算法，其原理就是：在一段时间间隔内，对请求进行计数，与阀值进行比较判断是否需要限流，每结束一个时间间隔时，都会将计数器清零。
计数器算法存在“时间临界点”缺陷，比如每一分钟限制100个请求，可以在00:00:00-00:00:58秒里面都没有请求，在00:00:59瞬间发送100个请求，这个对于计数器算法来是允许的，然后在00:01:00再次发送100个请求，意味着在短短1s内发送了200个请求，如果量更大呢，系统可能会承受不住瞬间流量，导致系统崩溃。
滑动窗口算法的思想主要来源于Tcp协议，即将一个大的时间窗口分成多个小窗口，每次大窗口向后滑动一个小窗口，并保证大的窗口内流量不会超出最大值，这种实现比固定窗口的流量曲线更加平滑。
对于上述计数器算法存在的不足，对于滑动时间窗口，我们可以把1s的时间窗口划分成10个小窗口，或者想象窗口有10个时间片, 每个时间片统计自己片内100ms的请求数量。每过100ms，都有一个新的时间片加入窗口，早于当前时间1s的时间片滑出窗口，窗口内最多维护10个时间片。
虽然滑动窗口的使用，可以对时间临界问题有一定的优化，但从根本上而言，其实该算法并没有真正地解决固定窗口算法的临界突发流量问题。
在介绍漏桶算法前，我们可以回忆一下，一个小学数学上的经典问题，有一个水池，有一个进水口，一个出水口，进水速度为x，出水速度为y，问多久可以把水池注满，相信这个问题大家都不陌生。简单来说漏桶算法，就是这道小学数学题，但是我们并不需要求解，首先想象有一个木桶，桶的容量是固定的。当有请求到来时先放到木桶中，处理请求时则以固定的速度从木桶中取出请求即可，如果木桶已经满了，可以直接返回请求频率超限的错误或进行一些其他处理，算法的好处是，流量一直是以一种均匀的速度在被消费，十分适合某些业务场景。
因为流入请求的速率是不固定的，但流出的速率是恒定的，所以当系统面对突发流量时会有大量的请求失败。这就导致，在类似于电商抢购、微博热点、过年红包等场景中，该算法并不适用。
令牌桶有点像反方向的"漏桶"，它是以恒定的速度往木桶里加入令牌，木桶满了则不再加入令牌。服务收到请求时尝试从木桶中取出一个令牌，如果能够得到令牌则继续执行后续的业务逻辑。如果没有得到令牌，直接返回请求频率超限的错误或进行一些其他处理，不继续执行后续的业务逻辑。
同时由于向木桶中添加令牌的速度是恒定的，且木桶的容量是有上限等，所以单位时间内系统能够处理的请求数量也是可控的，从而起到限流的目的。假设加入令牌的速度为 100/s，桶的容量为500，那么在请求比较的少的时候，木桶可以先积攒一些令牌（最多500个）。当有突发流量时，可以一下把木桶内的令牌都取出，也就是500的并发，之后就需要等到有新的令牌加入后，才可以进行新的业务处理了。
golang官方提供的扩展库里就自带了限流算法的实现，即 golang.org/x/time/rate ,该限流器也是基于令牌桶实现的。time/rate包中使用 Limiter 类型对限流器进行了定义，所有限流功能都是基于 Limiter 实现的，其结构如下：
type Limiter struct { mu sync.Mutex limit Limit burst int // 令牌桶的大小 tokens float64 last time.Time // 上次更新tokens的时间 lastEvent time.Time // 上次发生限速器事件的时间（通过或者限制都是限速器事件）}代码示例：
package mainimport ( "fmt" "time" "golang.org/x/net/context" "golang.org/x/time/rate")func main() { wait() allow() release()}func wait() { limiter := rate.NewLimiter(10, 100) fmt.Println(limiter.Limit(), limiter.Burst()) c, _ := context.WithCancel(context.TODO()) for { limiter.Wait(c) time.Sleep(200 * time.Millisecond) fmt.Println(time.Now().Format("2016-01-02 15:04:05.000")) }}func allow() { limit := rate.Every(100 * time.Millisecond) limiter := rate.NewLimiter(limit, 10) for { if limiter.AllowN(time.Now(), 2) { fmt.Println(time.Now().Format("2016-01-02 15:04:05.000")) } else { time.Sleep(6 * time.Second) } }}func release() { limiter := rate.NewLimiter(10, 100) for { r := limiter.ReserveN(time.Now(), 20) time.Sleep(r.Delay()) fmt.Println(time.Now().Format("2016-01-02 15:04:05.000")) }}rate.NewLimiter 有两个参数，第一个参数是 r Limit，设置的是限流器Limiter的limit字段，代表每秒可以向桶中放入多少个 token，第二个参数是 b int，b 代表桶的大小，也就是限流器 Limiter 的burst字段。
对于 wait() 例子而言，其构造出的限流器的令牌桶大小为100,并以每秒10个Token的速率向桶中放置Token。同时除了给 r Limit 参数直接指定每秒产生的 Token 个数外，还可以使用 Every 方法来指定向桶中放置 Token 的时间间隔，如例子allow()，代表每 100ms 向桶中放入一个 Token。
Limiter 提供了三类方法供程序消费 Token，可以每次消费一个 Token，也可以一次性消费多个 Token。每种方法代表了当 Token 不足时，各自不同的对应手段，可以阻塞等待桶中Token补充，也可以直接返回取Token失败。
func (lim *Limiter) Wait(ctx context.Context) (err error)func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)Wait 实际上就是 WaitN(ctx,1)，当使用 Wait 方法消费 Token 时，如果此时桶内 Token 数组不足 (小于 N)，那么 Wait 方法将会阻塞一段时间，直至 Token 满足条件，若 Token 数量充足则直接返回。这里可以看到，Wait 方法有一个 context 参数。我们可以设置 context 的 Deadline 或者 Timeout，来决定此次 Wait 的最长时间，具体可以参考示例 wait()。
func (lim *Limiter) Allow() boolfunc (lim *Limiter) AllowN(now time.Time, n int) boolAllow 实际上就是 AllowN(time.Now(),1)，AllowN 方法表示，截止到某一时刻，目前桶中 token 数目是否大于 n 个，若满足则返回 true，同时从桶中消费 n 个 token。若不满足，直接返回false。
func (lim *Limiter) Reserve() *Reservationfunc (lim *Limiter) ReserveN(now time.Time, n int) *ReservationReserve 相当于 ReserveN(time.Now(),1)，ReserveN 的用法相对复杂一些，当调用完成后，无论 Token 是否充足，都会返回一个 Reservation 的指针对象。我们可以通过调用该对象的Delay()方法得到一个需要等待的时间，时间到后即可进行业务处理。若不想等待，则可以调用Cancel()方法，该方法会将 Token 归还。
Limiter 支持创建后动态调整速率和桶的大小，我们可以根据现有环境和条件的不同，动态地改变 Token 桶大小和生成速率。
func (lim *Limiter) SetLimit(newLimit Limit) {}func (lim *Limiter) SetBurst(newBurst int) {}uber 在 Github 上开源了一套用于服务限流的 go 语言库 ratelimit, 该组件是基于 Leaky Bucket(漏桶)实现的，github地址：
https://github.com/uber-go/ratelimit
package mainimport ( "fmt" "go.uber.org/ratelimit" "time")func main() { // 每秒可以通过100个请求，也就是平均每个请求间隔10ms rl := ratelimit.New(100) prev := time.Now() for i := 0; i < 10; i++ { now := rl.Take() fmt.Println(i, now.Sub(prev)) prev = now }}相比于传统漏桶算法，该算法提出了一个最大松弛量的概念，以应对线上流量激增的场景，即请求 1 完成后，15ms 后，请求 2 才到来，可以对请求 2 立即处理。请求 2 完成后，5ms 后，请求 3 到来，这个时候距离上次请求还不足 10ms，因此还需要等待 5ms。但是，对于这种情况，实际上三个请求一共消耗了 25ms 才完成，并不是预期的 20ms。在 uber-go 实现的 ratelimit 中，可以把之前间隔比较长的请求的时间，匀给后面的使用，保证每秒请求数 (RPS) 。对于上述 case，因为请求 2 相当于多等了 5ms，我们可以把这 5ms 移给请求 3 使用。加上请求 3 本身就是 5ms 之后过来的，一共刚好 10ms，所以请求 3 无需等待，直接可以处理。此时三个请求也恰好一共是 20ms。
t.sleepFor += t.perRequest - now.Sub(t.last)if t.sleepFor > 0 { t.clock.Sleep(t.sleepFor) t.last = now.Add(t.sleepFor) t.sleepFor = 0} else { t.last = now}当 t.sleepFor > 0，代表此前的请求多余出来的时间，无法完全抵消此次的所需量，因此需要 sleep 相应时间, 同时将 t.sleepFor 置为 0。当 t.sleepFor < 0，说明此次请求间隔大于预期间隔，将多出来的时间累加到 t.sleepFor 即可。但是，对于某种情况，请求 1 完成后，请求 2 过了很久到达 (好几个小时都有可能)，那么此时对于请求 2 的请求间隔 now.Sub(t.last)，会非常大。以至于即使后面大量请求瞬时到达，也无法抵消完这个时间，那这样就失去了限流的意义。为了防止这种情况，ratelimit 就引入了最大松弛量 (maxSlack) 的概念, 该值为负值，表示允许抵消的最长时间，防止以上情况的出现。maxSlack 的值为 -10 * time.Second / time.Duration(rate), 即是十个请求的间隔大小。我们也可以理解为 ratelimit 允许的最大瞬时请求为 10。
func New(rate int, opts ...Option) Limiter初始化限流器时，方法支持可选参数opts，常用的 Option 为 WithoutSlack，设置该配置后，限流器将取消最大松弛量的优化，严格按照间隔时间处理请求，WithClock 支持自定义时钟，使限流器使用用户自定义时钟进行时间处理。
Hystrix的golang版本项目地址是：
https://github.com/afex/hystrix-go
Hystrix是Netflix开源的一个限流熔断的项目、主要有以下功能:
配置说明
type CommandConfig struct{ Timeout int // 执行command的超时时间为3s MaxConcurrentRequests int // command的最大并发量 RequestVolumeThreshold int // 统计窗口10s内的请求数量，达到这个请求数量后才去判断是否要开启熔断 SleepWindow int // 当熔断器被打开后，SleepWindow的时间就是控制过多久后去尝试服务是否可用了 ErrorPercentThreshold int // 错误百分比，请求数量大于等于RequestVolumeThreshold并且错误率到达这个百分比后就会启动熔断}字段 | 说明 |
Timeout | 执行command的超时时间，默认时间是1000毫秒 |
MaxConcurrentRequests | command的最大并发量，默认值是10 |
SleepWindow | 当熔断器被打开后，SleepWindow的时间就是控制过多久后去尝试服务是否可用了，默认值是5000毫秒 |
RequestVolumeThreshold | 一个统计窗口10秒内请求数量，达到这个请求数量后才去判断是否要开启熔断，默认值是20 |
ErrorPercentThreshold | 错误百分比，请求数量大于等于RequestVolumeThreshold并且错误率到达这个百分比后就会启动熔断，默认值是50 |
主要函数
func Go(name string, run runFunc, fallback fallbackFunc)func GoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) func Do(name string, run runFunc, fallback fallbackFunc)func DoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC)Do内部调用的Doc方法，Go内部调用的是Goc方法，在Doc方法内部最终调用的还是Goc方法，只是在Doc方法内做了同步逻辑
其中name为限流器名称，初始化时传递
run为业务逻辑函数，其结构为func() error
fallback为失败回调函数，其结构为func(err error) error
func DoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) error { ..... 省略部分封装代码 var errChan chan error if fallback == nil { errChan = GoC(ctx, name, r, nil) } else { errChan = GoC(ctx, name, r, f) } select { case <-done: return nil case err := <-errChan: return err }}使用示例
package mainimport ( "errors" "fmt" "github.com/afex/hystrix-go/hystrix" "net/http" "sync" "testing" "time")func TestHystrixRequestWithErr(t *testing.T) { hystrix.ConfigureCommand("mycommand", hystrix.CommandConfig{ Timeout: 1000, // 超时时间1秒 MaxConcurrentRequests: 20, // 最大并发数量20 SleepWindow: 1000, // 窗口时间1秒，熔断开启1秒后尝试重试 RequestVolumeThreshold: 5, // 10秒钟请求数量超过5次，启动熔断器判断 ErrorPercentThreshold: 50, // 请求数超过5并且错误率达到百分之50，开启熔断 }) wg := new(sync.WaitGroup) // 模拟并发10次请求，5次返回err，导致熔断器开启 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } if i%2 == 0 { return errors.New("测试错误!") } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为开启状态 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 睡眠1秒，转换为半开状态，并发请求10次，应该会有一个goroutine真正去请求，返回成功，其它请求直接走fallback逻辑 time.Sleep(1 * time.Second) for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 熔断器已经由半开转为关闭状态，请求应该全部成功 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------")}func TestHystrixRequestWithOverLimit(t *testing.T) { hystrix.ConfigureCommand("mycommand", hystrix.CommandConfig{ Timeout: 1000, // 超时时间1秒 MaxConcurrentRequests: 20, // 最大并发数量20 SleepWindow: 10000, // 窗口时间1秒，熔断开启10秒后尝试重试 RequestVolumeThreshold: 5, // 10秒钟请求数量超过5次，启动熔断器判断 ErrorPercentThreshold: 50, // 请求数超过5并且错误率达到百分之50，开启熔断 }) wg := new(sync.WaitGroup) // 模拟并发大于20 for i := 0; i < 40; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为开启状态 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为半开启状态 time.Sleep(10 * time.Second) for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为关闭状态 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------")}func TestHystrixRequestWithTimeout(t *testing.T) { hystrix.ConfigureCommand("mycommand", hystrix.CommandConfig{ Timeout: 1000, // 超时时间1秒 MaxConcurrentRequests: 20, // 最大并发数量20 SleepWindow: 10000, // 窗口时间10秒，熔断开启1秒后尝试重试 RequestVolumeThreshold: 5, // 10秒钟请求数量超过5次，启动熔断器判断 ErrorPercentThreshold: 50, // 请求数超过5并且错误率达到百分之50，开启熔断 }) wg := new(sync.WaitGroup) // 模拟请求超时 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } if i%2 == 0 { time.Sleep(1 * time.Second) return nil } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为开启状态 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为半开启状态 time.Sleep(10 * time.Second) for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------") // 继续模拟10次请求，熔断器应该为关闭状态 for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _ = hystrix.Do("mycommand", func() error { _, err := http.Get("https://baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return err } fmt.Println("success!") return nil }, func(err error) error { fmt.Printf("handle error:%v\n", err) return nil }) }(i) } wg.Wait() fmt.Println("----------------")}项目地址为：
https://github.com/sony/gobreaker
gobreaker是索尼的开源的一个限流熔断的项目，是基于《微软云设计模式》一书中的熔断器模式的 Golang 实现的，本质利用的还是原子计数法、主要有以下功能:
配置说明
type Settings struct { Name string MaxRequests uint32 Interval time.Duration Timeout time.Duration ReadyToTrip func(counts Counts) bool OnStateChange func(name string, from State, to State)}字段 | 说明 |
MaxRequests | 最大请求数，当在最大请求数下，均请求正常的情况下，会关闭熔断器 |
Interval | 一个正常的统计周期，如果为0，那每次都会将计数清零 |
Timeout | 进入熔断后，可以再次请求的时间 |
ReadyToTrip | 判断熔断生效的钩子函数(通过Counts判断是否开启熔断，默认连续失败超过5次) |
OnStateChange | 状态变更的钩子函数(看是否需要) |
使用示例
package mainimport ( "errors" "fmt" "net/http" "sync" "testing" "time" "github.com/sony/gobreaker")func TestGoBreaker(t *testing.T) { cb := gobreaker.NewCircuitBreaker(gobreaker.Settings{ Name: "Get", MaxRequests: 10, Interval: time.Second * 2, Timeout: time.Second * 2, }) for i := 0; i < 10; i++ { _, err := cb.Execute(func() (interface{}, error) { if i < 6 { return nil, errors.New("测试错误!") } fmt.Println("success!") return nil, nil }) if err != nil { fmt.Println(err) } } fmt.Println("--------------------") wg := new(sync.WaitGroup) for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _, err := cb.Execute(func() (interface{}, error) { _, err := http.Get("https://www.baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return nil, err } fmt.Println("success!") return nil, nil }) if err != nil { fmt.Println(err) } }(i) } wg.Wait() fmt.Println("--------------------") time.Sleep(1 * time.Second) for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _, err := cb.Execute(func() (interface{}, error) { _, err := http.Get("https://www.baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return nil, err } fmt.Println("success!") return nil, nil }) if err != nil { fmt.Println(err) } }(i) } wg.Wait() fmt.Println("--------------------") time.Sleep(1 * time.Second) for i := 0; i < 10; i++ { wg.Add(1) go func(i int) { defer wg.Done() _, err := cb.Execute(func() (interface{}, error) { _, err := http.Get("https://www.baidu.com") if err != nil { fmt.Printf("请求失败, err:%s\n", err.Error()) return nil, err } fmt.Println("success!") return nil, nil }) if err != nil { fmt.Println(err) } }(i) } wg.Wait() fmt.Println("--------------------")}
作者：萌新程序猿
链接：
https://www.jianshu.com/p/be5c139c11e3
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
声明：本站内容部分源于网络转载，出于传递更多信息之目的，并不意味着赞同其观点或证实其描述。文章内容仅供参考，请咨询相关专业人士。
如果无意之中侵犯了您的版权，或有意见、反馈或投诉等情况, 请联系本站，[qq:]
Copyright ©2024 编程密语 All rights reserved 版权所有 鲁ICP备09004228号-12
鲁公网安备37020202000738号