六.Go微服务--熔断

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

1. 前言

无论是令牌桶, 漏桶
还是 自适应限流的方法，总的来说都是服务端的单机限流方式。虽然服务端限流虽然可以帮助我们抗住一定的压力，但是拒绝请求毕竟还是有成本的。如果我们的本来流量可以支撑 1w rps，加了限流可以支撑在 10w rps 的情况下仍然可以提供 1w rps 的有效请求，但是流量突然再翻了 10 倍，来到 100w rps 那么服务该挂还是得挂。

所以我们的可用性建设不仅仅是服务端做建设就可以万事大吉了，得在整个链路上的每个组件都做好自己的事情才行，今天我们就来一起看一下客户端上的限流措施：熔断。

2.熔断器

如上图所示，熔断器存在三个状态:

关闭(closed): 关闭状态下没有触发断路保护，所有的请求都正常通行
打开(open): 当错误阈值触发之后，就进入开启状态，这个时候所有的流量都会被节流，不运行通行
半打开(half-open): 处于打开状态一段时间之后，会尝试尝试放行一个流量来探测当前 server 端是否可以接收新流量，如果这个没有问题就会进入关闭状态，如果有问题又会回到打开状态

2.1 hystrix-go

熔断器中比较典型的实现就是 hystrix，Golang 也有对应的版本，我们先来看一下 hystrix-go 是怎么实现的

2.2 案例

先看一个使用案例，首先我们使用 gin 启动一个服务端，这个服务端主要是前 200ms 的请求都会返回 500，之后的请求都会返回 200

func server() {
	e := gin.Default()
	start := time.Now.Unix()
	
	e.GET("/ping", func(ctx *gin.Context) {
		if time.Since(start) < 201*time.Millisecond {
			ctx.String(http.StatusInternalServerError, "pong")
			return
		}
		ctx.String(http.StatusOK, "pong")
	})
	e.Run(":8080")
}

然后配置 hystrix，hystrix.ConfigureCommand(command name, config)hystrix 的配置是按照每个 command 进行配置，使用的时候我们也需要传递一个 command，下面的配置就是我们的请求数量大于等于 10 个并且错误率大于等于 20% 的时候就会触发熔断器开关，熔断器打开 500ms 之后会进入半打开的状态，尝试放一部分请求去访问

func main(){
	hystrix.ConfigureCommand("test", hystrix.CommandConfig{
		// 执行 command 的超时时间
		Timeout: 10,

		// 最大并发量
		MaxConcurrentRequests: 100,

		// 一个统计窗口 10 秒内请求数量
		// 达到这个请求数量后才去判断是否要开启熔断
		RequestVolumeThreshold: 10,

		// 熔断器被打开后
		// SleepWindow 的时间就是控制过多久后去尝试服务是否可用了
        // 单位为毫秒
		SleepWindow: 500,

		// 错误百分比
		// 请求数量大于等于 RequestVolumeThreshold 并且错误率到达这个百分比后就会启动熔断
		ErrorPercentThreshold: 20,
	})
}

然后我们使用一个循环当做客户端代码，会请求 20 次，每一个请求消耗 100ms

func main() {
	go server()
	
  // 这里是 config 代码

	for i := 0; i < 20; i++ {
		_ = hystrix.Do("test", func() error {
			resp, _ := resty.New().R().Get("http://localhost:8080/ping")
			if resp.IsError() {
				return fmt.Errorf("err code: %s", resp.Status())
			}
			return nil
		}, func(err error) error {
			fmt.Println("fallback err: ", err)
			return err
		})
		// 每次请求间隔100ms
		time.Sleep(100 * time.Millisecond)
	}

}

完整代码如下：

package main

import (
	"fmt"
	"net/http"
	"time"

	"github.com/afex/hystrix-go/hystrix"
	"github.com/gin-gonic/gin"
	"gopkg.in/resty.v1"
)

func server() {
	e := gin.Default()
	start := time.Now()
	e.GET("/ping", func(ctx *gin.Context) {
		if time.Since(start) < 201*time.Millisecond {
			ctx.String(http.StatusInternalServerError, "pong")
			return
		}
		ctx.String(http.StatusOK, "pong")
	})

	err := e.Run(":8080")
	if err != nil {
		fmt.Printf("START SERVER OCCUR ERROR, %s", err.Error())
	}
}

func main() {
	go server()

	hystrix.ConfigureCommand("test", hystrix.CommandConfig{
		// 执行 command 的超时时间
		Timeout: 10,

		// 最大并发量
		MaxConcurrentRequests: 100,

		// 一个统计窗口 10 秒内请求数量
		// 达到这个请求数量后才去判断是否要开启熔断
		RequestVolumeThreshold: 10,

		// 熔断器被打开后
		// SleepWindow 的时间就是控制过多久后去尝试服务是否可用了
		// 单位为毫秒
		SleepWindow: 500,

		// 错误百分比
		// 请求数量大于等于 RequestVolumeThreshold 并且错误率到达这个百分比后就会启动熔断
		ErrorPercentThreshold: 20,
	})

	// 模拟20个客户端请求
	for i := 0; i < 20; i++ {
		_ = hystrix.Do("test", func() error {
			resp, _ := resty.New().R().Get("http://localhost:8080/ping")
			if resp.IsError() {
				return fmt.Errorf("err code: %s", resp.Status())
			}
			return nil
		}, func(err error) error {
			fmt.Println("fallback err: ", err)
			return err
		})
		time.Sleep(100 * time.Millisecond)
	}

}

执行的结果如下，前面 2 个请求报 500，等到发起了 10 个请求之后就会进入熔断， 500ms 也就是发出 5 个请求之后就会重新去请求服务端

2.3 hystrix-go 核心实现

核心实现的方法是 AllowRequest，IsOpen判断当前是否处于熔断状态，allowSingleTest就是去看是否过了一段时间需要重新进行尝试

func (circuit *CircuitBreaker) AllowRequest() bool {
    // 判断是否打开或者过了一段时间需要重新进行尝试
	return !circuit.IsOpen() || circuit.allowSingleTest()
}

IsOpen先看当前是否已经打开了，如果已经打开了就直接返回就行了，如果还没打开就去判断

请求数量是否满足要求
请求的错误率是否过高，如果两个都满足就会打开熔断器

func (circuit *CircuitBreaker) IsOpen() bool {
	circuit.mutex.RLock()
	// 查看打开状态
	o := circuit.forceOpen || circuit.open
	circuit.mutex.RUnlock()

	if o {
		return true
	}
    
    // 获取定期时间内请求的数量是否小于 设定的请求断融数量，如果小于则不开启断融
	if uint64(circuit.metrics.Requests().Sum(time.Now())) < getSettings(circuit.Name).RequestVolumeThreshold {
		return false
	}
    
    // 开启断融前判断服务是否健康
	if !circuit.metrics.IsHealthy(time.Now()) {
		// too many failures, open the circuit
		circuit.setOpen()
		return true
	}

	return false
}

hystrix-go已经可以比较好的满足我们的需求，但是存在一个问题就是一旦触发了熔断，在一段时间之类就会被一刀切的拦截请求，所以我们来看看 google sre 的一个实现

2.4 GoogleSre 过载保护算法

$$
max(0, \frac{requests - K * accepts}{requests + 1})
$$

算法如上所示，这个公式计算的是请求被丢弃的概率

requests: 一段时间的请求数量
accepts: 成功的请求数量
K：倍率，K越小表示越激进，越小表示越容易被丢弃请求

这个算法的好处是不会直接一刀切的丢弃所有请求，而是计算出一个概率来进行判断，当成功的请求数量越少，K越小的时候 $requests - K * accepts$ 的值就越大，计算出的概率也就越大，表示这个请求被丢弃的概率越大

2.5 Kratos 实现分析

核心代码

func (b *sreBreaker) Allow() error {
	// 统计成功的请求，和总的请求
	success, total := b.summary()

	// 计算当前的成功率
	k := b.k * float64(success)
	if log.V(5) {
		log.Info("breaker: request: %d, succee: %d, fail: %d", total, success, total-success)
	}
	// 统计请求量和成功率
	// 如果 rps 比较小，不触发熔断
	// 如果成功率比较高，不触发熔断，如果 k = 2，那么就是成功率 >= 50% 的时候就不熔断
	if total < b.request || float64(total) < k {
		if atomic.LoadInt32(&b.state) == StateOpen {
			atomic.CompareAndSwapInt32(&b.state, StateOpen, StateClosed)
		}
		return nil
	}
	if atomic.LoadInt32(&b.state) == StateClosed {
		atomic.CompareAndSwapInt32(&b.state, StateClosed, StateOpen)
	}

	// 计算一个概率，当 dr 值越大，那么被丢弃的概率也就越大
	// dr 值是，如果失败率越高或者是 k 值越小，那么它越大
	dr := math.Max(0, (float64(total)-k)/float64(total+1))
	drop := b.trueOnProba(dr)
	if log.V(5) {
		log.Info("breaker: drop ratio: %f, drop: %t", dr, drop)
	}
	if drop {
		return ecode.ServiceUnavailable
	}
	return nil
}

// 通过随机来判断是否需要进行熔断
func (b *sreBreaker) trueOnProba(proba float64) (truth bool) {
	b.randLock.Lock()
	truth = b.r.Float64() < proba
	b.randLock.Unlock()
	return
}

3. 总结

可用性仅靠服务端来保证是不靠谱的，只有整条链路上的所有服务都做好了自己可用性相关的建设我们的服务 SLA 最后才能够有保证。

最常见的hystrix-go 和 kratos 两种熔断方式，kratos采用 Google SRE 的实现的好处就是没有半开的状态，也没有完全开启的状态，而是通过一个概率来进行判断我们的流量是否应该通过，这样没有那么死板，也可以保证我们错误率比较高的时候不会大量请求服务端，给服务端喘息恢复的时间。