hystrix-go 使用與原理

開(kāi)篇

這周在看內(nèi)部一個(gè)熔斷限流包時(shí)，發(fā)現(xiàn)它是基于一個(gè)開(kāi)源項(xiàng)目?hystrix-go?實(shí)現(xiàn)了，隨即看了兩天的源碼，因此有了這篇文章。

Hystrix

Hystrix?是由?Netflex?開(kāi)發(fā)的一款開(kāi)源組件，提供了基礎(chǔ)的熔斷功能。?Hystrix?將降級(jí)的策略封裝在?Command?中，提供了?run?和?fallback?兩個(gè)方法，前者表示正常的邏輯，比如微服務(wù)之間的調(diào)用……，如果發(fā)生了故障，再執(zhí)行?fallback?方法返回結(jié)果，我們可以把它理解成保底操作。如果正常邏輯在短時(shí)間內(nèi)頻繁發(fā)生故障，那么可能會(huì)觸發(fā)斷路，也就是之后的請(qǐng)求不再執(zhí)行?run, 而是直接執(zhí)行?fallback。更多關(guān)于?Hystrix?的信息可以查看?https://github.com/Netflix/Hystrix，而
hystrix-go?則是用?go?實(shí)現(xiàn)的?hystrix?版，更確切的說(shuō)，是簡(jiǎn)化版。只是上一次更新還是 2018 年 的一次?pr, 也就畢業(yè)了？

為什么需要這些工具？

比如一個(gè)微服務(wù)化的產(chǎn)品線上，每一個(gè)服務(wù)都專注于自己的業(yè)務(wù)，并對(duì)外提供相應(yīng)的服務(wù)接口，或者依賴于外部服務(wù)的某個(gè)邏輯接口，就像下面這樣。

假設(shè)我們當(dāng)前是?服務(wù)A，有部分邏輯依賴于?服務(wù)C，服務(wù)C?又依賴于?服務(wù)E, 當(dāng)前微服務(wù)之間進(jìn)行?rpc?或者?http?通信，假設(shè)此時(shí)?服務(wù)C?調(diào)用 服務(wù) E 失敗，比如由于網(wǎng)絡(luò)波動(dòng)導(dǎo)致超時(shí)或者服務(wù) E 由于過(guò)載，服務(wù)E 已經(jīng) down 掉了。

調(diào)用失敗，一般會(huì)有失敗重試等機(jī)制。但是再想想，假設(shè)服務(wù) E 已然不可用的情況下，此時(shí)新的調(diào)用不斷產(chǎn)生，同時(shí)伴隨著調(diào)用等待和失敗重試，會(huì)導(dǎo)致 服務(wù) C 對(duì)服務(wù) E 的調(diào)用而產(chǎn)生大量的積壓，慢慢會(huì)耗盡服務(wù) C 的資源，進(jìn)而導(dǎo)致服務(wù) C 也 down 掉，這樣惡性循環(huán)下，會(huì)影響到整個(gè)微服務(wù)體系，產(chǎn)生雪崩效應(yīng)。

雖然導(dǎo)致雪崩的發(fā)生不僅僅這一種，但是我們需要采取一定的措施，來(lái)保證不讓這個(gè)噩夢(mèng)發(fā)生。而?hystrix-go?就很好的提供了 熔斷和降級(jí)的措施。它的主要思想在于，設(shè)置一些閥值，比如最大并發(fā)數(shù)(當(dāng)并發(fā)數(shù)大于設(shè)置的并發(fā)數(shù)，攔截)。錯(cuò)誤率百分比(請(qǐng)求數(shù)量大于等于設(shè)置 的閥值，并且錯(cuò)誤率達(dá)到設(shè)置的百分比時(shí)，觸發(fā)熔斷)以及熔斷嘗試恢復(fù)時(shí)間等。

使用

hystrix-go?的使用非常簡(jiǎn)單，你可以調(diào)用它的?Go?或者?Do?方法，只是?Go?方法是異步的方式。而?Do?方法是同步方式。我們從一個(gè)簡(jiǎn)單的例子開(kāi)啟。

_ = hystrix.Do("wuqq", func() error {        // talk to other services        _, err := http.Get("https://www.baidu.com/")        if err != nil {            fmt.Println("get error:%v",err)            return err        }        return nil    }, func(err error) error {        fmt.Printf("handle  error:%v\n", err)        return nil    })

Do?函數(shù)需要三個(gè)參數(shù)，第一個(gè)參數(shù)?commmand?名稱，你可以把每個(gè)名稱當(dāng)成一個(gè)獨(dú)立當(dāng)服務(wù)，第二個(gè)參數(shù)是處理正常的邏輯，比如?http?調(diào)用服務(wù)，返回參數(shù)是?err。如果處理或者調(diào)用失敗，那么就執(zhí)行第三個(gè)參數(shù)邏輯， 我們稱為保底操作。假設(shè)服務(wù)錯(cuò)誤率過(guò)高而導(dǎo)致熔斷器開(kāi)啟，那么之后的請(qǐng)求也直接回調(diào)此函數(shù)。

既然熔斷器是按照配置的規(guī)則而進(jìn)行是否開(kāi)啟的操作，那么我們當(dāng)然可以設(shè)置我們想要的值。

hystrix.ConfigureCommand("wuqq", hystrix.CommandConfig{        Timeout:                int(3 * time.Second),        MaxConcurrentRequests:  10,        SleepWindow:            5000,        RequestVolumeThreshold: 10,        ErrorPercentThreshold:  30,    })    _ = hystrix.Do("wuqq", func() error {        // talk to other services        _, err := http.Get("https://www.baidu.com/")        if err != nil {            fmt.Println("get error:%v",err)            return err        }        return nil    }, func(err error) error {        fmt.Printf("handle  error:%v\n", err)        return nil    })

稍微解釋一下上面配置的值含義:

• Timeout: 執(zhí)行?command?的超時(shí)時(shí)間。

• MaxConcurrentRequests：command?的最大并發(fā)量 。

• SleepWindow：當(dāng)熔斷器被打開(kāi)后，SleepWindow?的時(shí)間就是控制過(guò)多久后去嘗試服務(wù)是否可用了。

• RequestVolumeThreshold：一個(gè)統(tǒng)計(jì)窗口 10 秒內(nèi)請(qǐng)求數(shù)量。達(dá)到這個(gè)請(qǐng)求數(shù)量后才去判斷是否要開(kāi)啟熔斷

• ErrorPercentThreshold：錯(cuò)誤百分比，請(qǐng)求數(shù)量大于等于?RequestVolumeThreshold?并且錯(cuò)誤率到達(dá)這個(gè)百分比后就會(huì)啟動(dòng)熔斷

當(dāng)然你不設(shè)置的話，那么自動(dòng)走的默認(rèn)值。

我們?cè)賮?lái)看一個(gè)簡(jiǎn)單的例子:

package main
import (   "fmt" "github.com/afex/hystrix-go/hystrix" "net/http" "time")
type Handle struct{}
func (h *Handle) ServeHTTP(r http.ResponseWriter, request *http.Request) {   h.Common(r, request)}
func (h *Handle) Common(r http.ResponseWriter, request *http.Request) {   hystrix.ConfigureCommand("mycommand", hystrix.CommandConfig{      Timeout:                int(3 * time.Second),      MaxConcurrentRequests:  10,      SleepWindow:            5000,      RequestVolumeThreshold: 20,      ErrorPercentThreshold:  30,   })   msg := "success"
  _ = hystrix.Do("mycommand", func() error {      _, err := http.Get("https://www.baidu.com")      if err != nil {         fmt.Printf("請(qǐng)求失敗:%v", err)         return err  }      return nil  }, func(err error) error {      fmt.Printf("handle  error:%v\n", err)      msg = "error"  return nil  })   r.Write([]byte(msg))}
func main() {   http.ListenAndServe(":8090", &Handle{})}

我們開(kāi)啟了一個(gè)?http?服務(wù)，監(jiān)聽(tīng)端口號(hào)?8090，所有請(qǐng)求的處理邏輯都在?Common?方法中，在這個(gè)方法中，我們主要是發(fā)起一次?http?請(qǐng)求，請(qǐng)求成功響應(yīng)?success, 如果失敗，響應(yīng)失敗原因。

我們?cè)賹?xiě)另一個(gè)簡(jiǎn)單程序，并發(fā)?11?次的請(qǐng)求?8090?端口。

package main
import (    "fmt"    "io/ioutil"    "net/http"    "sync"    "time")
var client *http.Client
func init() {    tr := &http.Transport{        MaxIdleConns:    100,        IdleConnTimeout: 1 * time.Second,    }    client = &http.Client{Transport: tr}}
type info struct {    Data interface{} `json:"data"`}
func main() {    var wg sync.WaitGroup
    for i := 0; i < 11; i++ {        wg.Add(1)        go func(int2 int) {            defer wg.Done()            req, err := http.NewRequest("GET", "http://localhost:8090", nil)            if err != nil {                fmt.Printf("初始化http客戶端處錯(cuò)誤:%v", err)                return            }            resp, err := client.Do(req)            if err != nil {                fmt.Printf("初始化http客戶端處錯(cuò)誤:%v", err)                return            }            defer resp.Body.Close()            nByte, err := ioutil.ReadAll(resp.Body)            if err != nil {                fmt.Printf("讀取http數(shù)據(jù)失敗:%v", err)                return            }            fmt.Printf("接收到到值:%v\n", string(nByte))        }(i)    }    wg.Wait()
    fmt.Printf("請(qǐng)求完畢\n")}

由于我們配置?MaxConcurrentRequests?為 10，那么意味著還有個(gè) g 請(qǐng)求會(huì)失敗:

和我們想的一樣。

接著我們把網(wǎng)絡(luò)斷開(kāi)，并發(fā)請(qǐng)求改成 10 次。再次運(yùn)行程序并發(fā)請(qǐng)求?8090?端口，此時(shí)由于網(wǎng)絡(luò)已關(guān)閉，導(dǎo)致請(qǐng)求百度失敗：

接著繼續(xù)請(qǐng)求：

熔斷器已開(kāi)啟，上面我們配置的?RequestVolumeThreshold?和?ErrorPercentThreshold?生效。

然后我們把網(wǎng)連上，五秒后 (SleepWindow?的值) 繼續(xù)并發(fā)調(diào)用，當(dāng)前熔斷器處于半開(kāi)的狀態(tài)，此時(shí)請(qǐng)求允許調(diào)用依賴，如果成功則關(guān)閉，失敗則繼續(xù)開(kāi)啟熔斷器。

可以看到，有一個(gè)成功了，那么此時(shí)熔斷器已關(guān)閉，接下來(lái)繼續(xù)運(yùn)行函數(shù)并發(fā)調(diào)用：

可以看到，10 個(gè)都已經(jīng)是正常成功的狀態(tài)了。

那么問(wèn)題來(lái)了，為什么最上面的圖只有一個(gè)是成功的？5 秒已經(jīng)過(guò)了，并且當(dāng)前網(wǎng)絡(luò)正常，應(yīng)該是 10 個(gè)請(qǐng)求都成功，但是我們看到的只有一個(gè)是成功狀態(tài)。通過(guò)源碼我們可以找到答案:
具體邏輯在判斷當(dāng)前請(qǐng)求是否可以調(diào)用 ? ? ? ?

這段代碼首先判斷了熔斷器是否開(kāi)啟，并且當(dāng)前時(shí)間大于 上一次開(kāi)啟熔斷器的時(shí)間 +?SleepWindow?的時(shí)間，如果條件都符合的話，更新此熔斷器最新的?openedOrLastTestedTime?, 是通過(guò)?CompareAndSwapInt64?原子操作完成的，意味著必然只會(huì)有一個(gè)成功。

此時(shí)熔斷器還是半開(kāi)的狀態(tài)，接著如果能拿到令牌，執(zhí)行?run?函數(shù)（也就是 Do 傳入的第二個(gè)簡(jiǎn)單封裝后的函數(shù)），發(fā)起?http?請(qǐng)求，如果成功，上報(bào)成功狀態(tài)，關(guān)閉熔斷器。如果失敗，那么熔斷器依舊開(kāi)啟。

以上就是大體的流程講解，本來(lái)想當(dāng)一篇文章搞定，但是我又想好好梳理源碼，再寫(xiě)下去文章太長(zhǎng)了，因此花兩篇文章來(lái)介紹，下一篇純講源碼部分。

參考:https://www.cnblogs.com/li-peng/p/10997140.html。

推薦閱讀

• 防雪崩利器 hystrix-go 源碼分析