五、go-kit微服務熔斷機制

介紹

go-kit 提供了三種熔斷

1、 gobreaker

2、 handy

3、 hystrix-go

hystrix在java中用的比較多，我們來介紹下go-kit中hystrix的使用方法

go-kit的hystrix

Middleware的實作

1、 Hystrix傳回Middleware 此中間件會在原來的endPoint包一層Hystrix的endPoint

2、 hystrix通過傳入的commanName擷取對應的Hystrix的設定，并設定run失敗時運作的fallback函數為nil

3、 我們也可以自己實作middleware包裝endPoint

func Hystrix(commandName string) endpoint.Middleware {  
   return func(next endpoint.Endpoint) endpoint.Endpoint {  
      return func(ctx context.Context, request interface{}) (response interface{}, err error) {  
         var resp interface{}  
         if err := hystrix.Do(commandName, func() (err error) {  
            resp, err = next(ctx, request)  
            return err  
         }, nil); err != nil {  
            return nil, err  
         }  
         return resp, nil  
      }  
   }  
}
           

用戶端hystrix配置

1、Timeout 【請求逾時的時間】

2、ErrorPercentThreshold【允許出現的錯誤比例】

3、SleepWindow【熔斷開啟多久嘗試發起一次請求】

4、MaxConcurrentRequests【允許的最大并發請求數】

5、RequestVolumeThreshold 【波動期内的最小請求數，預設波動期10S】

commandName := "my-endpoint"  
hystrix.ConfigureCommand(commandName, hystrix.CommandConfig{  
  Timeout: *,  
  ErrorPercentThreshold:,  
  SleepWindow:,  
  MaxConcurrentRequests:,  
  RequestVolumeThreshold:,  
})
           

增加熔斷中間件的包裝

breakerMw := circuitbreaker.Hystrix(commandName)
//增加熔斷中間件  
reqEndPoint = breakerMw(reqEndPoint)
           

完整代碼

1、 關閉微服務的service

2、 在之前我們的用戶端代碼增加熔斷配置

3、 增加循環請求代碼

package main  

import (  
   "context"  
   "github.com/go-kit/kit/sd/etcdv3" 
   "time" 
   "github.com/go-kit/kit/sd" 
   "github.com/go-kit/kit/log" 
   "github.com/go-kit/kit/endpoint" 
   "io" 
   "github.com/go-kit/kit/sd/lb" 
   "grpc-test/pb" 
   "fmt" 
   "google.golang.org/grpc" 
   "github.com/afex/hystrix-go/hystrix" 
   "github.com/go-kit/kit/circuitbreaker"
)  

func main() {  
   commandName := "my-endpoint"  
   hystrix.ConfigureCommand(commandName, hystrix.CommandConfig{  
      Timeout: *,  
      ErrorPercentThreshold:,  
      SleepWindow:,  
      MaxConcurrentRequests:,  
      RequestVolumeThreshold:,  
   })  
   breakerMw := circuitbreaker.Hystrix(commandName)  


   var (  
      //注冊中心位址  
      etcdServer = "127.0.0.1:2379"  
      //監聽的服務字首  
      prefix     = "/services/book/"  
      ctx        = context.Background()  
   )  
   options := etcdv3.ClientOptions{  
      DialTimeout: time.Second *,  
      DialKeepAlive: time.Second *,  
   }  
   //連接配接注冊中心  
   client, err := etcdv3.NewClient(ctx, []string{etcdServer}, options)  
   if err != nil {  
      panic(err)  
   }  
   logger := log.NewNopLogger()  
   //建立執行個體管理器, 此管理器會Watch監聽etc中prefix的目錄變化更新緩存的服務執行個體資料  
   instancer, err := etcdv3.NewInstancer(client, prefix, logger)  
   if err != nil {  
      panic(err)  
   }  
   //建立端點管理器， 此管理器根據Factory和監聽的到執行個體建立endPoint并訂閱instancer的變化動态更新Factory建立的endPoint  
   endpointer := sd.NewEndpointer(instancer, reqFactory, logger)  
   //建立負載均衡器  
   balancer := lb.NewRoundRobin(endpointer)  

   /**  
   我們可以通過負載均衡器直接擷取請求的endPoint，發起請求  
   reqEndPoint,_ := balancer.Endpoint() 
   */  
   /**  
   也可以通過retry定義嘗試次數進行請求  
   */  
   reqEndPoint := lb.Retry,*time.Second, balancer)  

   //增加熔斷中間件  
   reqEndPoint = breakerMw(reqEndPoint)  
   //現在我們可以通過 endPoint 發起請求了  
   req := struct{}{}  
   for i :=; i <=; i++ {  
      if _, err = reqEndPoint(ctx, req); err != nil {  
         fmt.Println("目前時間: ", time.Now().Format("2006-01-02 15:04:05.99"))  
         fmt.Println(err)  
         time.sleep * time.Second)
      }  
   }  
}  


//通過傳入的 執行個體位址  建立對應的請求endPoint  
func reqFactory(instanceAddr string) (endpoint.Endpoint, io.Closer, error) {  
   return func(ctx context.Context, request interface{}) (interface{}, error) {  
      fmt.Println("請求服務: ", instanceAddr, "目前時間: ", time.Now().Format("2006-01-02 15:04:05.99"))  
      conn, err := grpc.Dial(instanceAddr, grpc.WithInsecure())  
      if err != nil {  
         fmt.Println(err)  
         panic("connect error")  
      }  
      defer conn.Close()  
      bookClient := book.NewBookServiceClient(conn)  
      bi,err := bookClient.GetBookInfo(context.Background(),&book.BookInfoParams{BookId})  
      fmt.Println(bi)  
      fmt.Println("       ", "擷取書籍詳情")  
      fmt.Println("       ", "bookId: 1", " => ", "bookName:", bi.BookName)  
      return nil,nil  
   },nil,nil  
}
           

輸出記錄

目前時間:  -- ::
no endpoints available (previously: no endpoints available; no endpoints available)
目前時間:  -- ::
no endpoints available (previously: no endpoints available; no endpoints available)
目前時間:  -- ::
no endpoints available (previously: no endpoints available; no endpoints available)
目前時間:  -- ::
no endpoints available (previously: no endpoints available; no endpoints available)
目前時間:  -- ::
no endpoints available (previously: no endpoints available; no endpoints available)
// :: hystrix-go: opening circuit my-endpoint
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
no endpoints available (previously: no endpoints available; no endpoints available)
// :: hystrix-go: allowing single test to possibly close circuit my-endpoint
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open
目前時間:  -- ::
hystrix: circuit open

Process finished with exit code 
           

通過上面的輸出記錄可以驗證我們的配置：

1、 前5條波動期内的錯誤，沒有觸發circuit開啟

2、 circuit開啟後請求熔斷生效

3、 circuit開啟10S後，SleepWindow測試發起請求設定生效