[关闭]
@aliasliyu4 2018-08-13T15:23:08.000000Z 字数 5970 阅读 3502

go-hystrix 源码读解

开局一张图
PgsQlq.md.png

文

下文是github上go-hystrix对与其功能的阐述，其脱胎于java版本的Hystrix库，主要目的是为了解决分布式系统中对于错误的保护，这一点从其熔断的定义也可以明白。

Hystrix is a latency and fault tolerance library designed to isolate points of access to remote systems, services and 3rd party libraries, stop cascading failure and enable resilience in complex distributed systems where failure is inevitable.

I think the Hystrix patterns of programmer-defined fallbacks and adaptive health monitoring are good for any distributed system. Go routines and channels are great concurrency primitives, but don't directly help our application stay available during failures.

hystrix-go aims to allow Go programmers to easily build applications with similar execution semantics of the Java-based Hystrix library.

command config定义

  1. type CommandConfig struct {
  2.     Timeout                int `json:"timeout"` // 超时时间定义
  3.     MaxConcurrentRequests  int `json:"max_concurrent_requests"` // 最大并发请求数
  4.     RequestVolumeThreshold int `json:"request_volume_threshold"` // 跳闸的最小请求数（不健康的断路器）
  5.     SleepWindow            int `json:"sleep_window"` // 跳闸之后可以重试的时间
  6.     ErrorPercentThreshold  int `json:"error_percent_threshold"` // 请求出错比
  7. }

断路器的定义

  1. // CircuitBreaker is created for each ExecutorPool to track whether requests
  2. // should be attempted, or rejected if the Health of the circuit is too low.
  3. type CircuitBreaker struct { // 断路器-定义
  4.     Name                   string // 名字
  5.     open                   bool   // 开启与否，关闭"open"=true，开启"open" = false
  6.     forceOpen              bool   // 强制开启
  7.     mutex                  *sync.RWMutex // 读写锁（unblock reading, block writer）
  8.     openedOrLastTestedTime int64 // 断路器被打开或者最近一次尝试的时间，尝试指断路器打开之后，系统探测是否可以发送请求。
  10.     executorPool *executorPool // 执行池
  11.     metrics      *metricExchange // 监控断路器
  12. }
  14. var (
  15.     circuitBreakersMutex *sync.RWMutex // 断路器锁
  16.     circuitBreakers      map[string]*CircuitBreaker // 注册断路器，所有的断路器都保存在这里
  17. )
  19. func init() {
  20.     circuitBreakersMutex = &sync.RWMutex{} // 初始化断路器锁
  21.     circuitBreakers = make(map[string]*CircuitBreaker) // 初始化断路器
  22. }

获取断路器

  1. // GetCircuit returns the circuit for the given command and whether this call created it.
  2. func GetCircuit(name string) (*CircuitBreaker, bool, error) {
  3.     circuitBreakersMutex.RLock() 
  4.     _, ok := circuitBreakers[name]
  5.     if !ok {
  6.         circuitBreakersMutex.RUnlock()
  7.         circuitBreakersMutex.Lock()
  8.         defer circuitBreakersMutex.Unlock() // 注意这里同时加了两次🔒且第二把锁是互斥锁，其中一个goroutine hold住并且赋值，锁释放。其他goroutine从内存中获取断路器
  9.         if cb, ok := circuitBreakers[name]; ok { // double check，防止其他的goroutine修改了全局变量circuitBreakers
  10.             return cb, false, nil
  11.         }
  12.         circuitBreakers[name] = newCircuitBreaker(name)
  13.     } else {
  14.         defer circuitBreakersMutex.RUnlock()
  15.     }
  17.     return circuitBreakers[name], !ok, nil
  18. }

执行池

  1. type executorPool struct { // 执行池
  2.     Name    string  // 名字
  3.     Metrics *poolMetrics // 执行池监控
  4.     Max     int // 最大的并发请求数量
  5.     Tickets chan *struct{} // 票证
  6. }
  8. // 开启一个新的执行池
  9. func newExecutorPool(name string) *executorPool {
  10.     p := &executorPool{}
  11.     p.Name = name // 名字
  12.     p.Metrics = newPoolMetrics(name) 
  13.     p.Max = getSettings(name).MaxConcurrentRequests // 从配置中获取最大的并发请求数量，如果配置中没有，则从默认配置中获取
  15.     p.Tickets = make(chan *struct{}, p.Max) // 初始化buffer chan
  16.     for i := 0; i < p.Max; i++ {
  17.         p.Tickets <- &struct{}{}
  18.     }
  19.     return p
  20. }
  22. func (p *executorPool) Return(ticket *struct{}) {
  23.     if ticket == nil {
  24.         return
  25.     }
  27.     p.Metrics.Updates <- poolMetricsUpdate{
  28.         activeCount: p.ActiveCount(),
  29.     }
  30.     p.Tickets <- ticket
  31. }
  33. func (p *executorPool) ActiveCount() int {
  34.     return p.Max - len(p.Tickets)
  35. }

执行函数入口GO，传递进去自定义的run，fallback函数，这个函数会封装ctx，然后继续调用GoC。

  1. func Go(name string, run runFunc, fallback fallbackFunc) chan error {
  2.     runC := func(ctx context.Context) error { // 匿名run带ctx
  3.         return run()
  4.     }
  5.     var fallbackC fallbackFuncC
  6.     if fallback != nil {
  7.         fallbackC = func(ctx context.Context, err error) error { // 匿名fallback带ctx
  8.             return fallback(err)
  9.         }
  10.     }
  11.     return GoC(context.Background(), name, runC, fallbackC)
  12. }

主要的执行逻辑GoC,提供了异步的执行。

  1. func GoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) chan error {
  2.     cmd := &command{ // command执行者
  3.         run:      run, // run
  4.         fallback: fallback, // fallback
  5.         start:    time.Now(), // 开始时间
  6.         errChan:  make(chan error, 1), // 错误
  7.         finished: make(chan bool, 1), // 是否完成
  8.     }
  10.     // dont have methods with explicit params and returns
  11.     // let data come in and out naturally, like with any closure
  12.     // explicit error return to give place for us to kill switch the operation (fallback)
  14.     circuit, _, err := GetCircuit(name) // 获取断路器
  15.     if err != nil {
  16.         cmd.errChan <- err
  17.         return cmd.errChan
  18.     }
  19.     cmd.circuit = circuit
  20.     ticketCond := sync.NewCond(cmd) // cond条件，具体使用见下文参考
  21.     ticketChecked := false
  23.     returnTicket := func() { // 
  24.         cmd.Lock()
  25.         // Avoid releasing before a ticket is acquired.
  26.         for !ticketChecked {
  27.             ticketCond.Wait() // 相当于select{}
  28.         }
  29.         cmd.circuit.executorPool.Return(cmd.ticket) // 将ticket放回池子中
  30.         cmd.Unlock()
  31.     }
  32.     returnOnce := &sync.Once{} // 确保被multi goroutine执行一次
  33.     reportAllEvent := func() { // events采集，后续dashboard使用
  34.         err := cmd.circuit.ReportEvent(cmd.events, cmd.start, cmd.runDuration)
  35.         if err != nil {
  36.             log.Printf(err.Error())
  37.         }
  38.     }
  40.     // g1, 检测断路器不允许通过，尝试fallback，将中途遇到的event上报。
  41.     go func() {
  42.         defer func() { cmd.finished <- true }()
  43.         if !cmd.circuit.AllowRequest() {
  44.             cmd.Lock()
  45.             // It's safe for another goroutine to go ahead releasing a nil ticket.
  46.             ticketChecked = true
  47.             ticketCond.Signal()
  48.             cmd.Unlock()
  49.             returnOnce.Do(func() {
  50.                 returnTicket()
  51.                 cmd.errorWithFallback(ctx, ErrCircuitOpen)
  52.                 reportAllEvent()
  53.             })
  54.             return
  55.         }
  58.         cmd.Lock()
  59.         select {
  60.         case cmd.ticket = <-circuit.executorPool.Tickets: // 从池子中取出ticket
  61.             ticketChecked = true
  62.             ticketCond.Signal() // 通知cond.Wait()
  63.             cmd.Unlock()
  64.         default:
  65.             ticketChecked = true
  66.             ticketCond.Signal()
  67.             cmd.Unlock()
  68.             returnOnce.Do(func() {
  69.                 returnTicket()
  70.                 cmd.errorWithFallback(ctx, ErrMaxConcurrency) // 并发过高
  71.                 reportAllEvent()
  72.             })
  73.             return
  74.         }
  76.         runStart := time.Now()
  77.         runErr := run(ctx)
  78.         returnOnce.Do(func() {
  79.             defer reportAllEvent()
  80.             cmd.runDuration = time.Since(runStart) // 运行时间
  81.             returnTicket() // 把ticket返回去
  82.             if runErr != nil {
  83.                 cmd.errorWithFallback(ctx, runErr) // fall back！
  84.                 return
  85.             }
  86.             cmd.reportEvent("success") // 执行成功
  87.         })
  88.     }()
  90.     go func() {
  91.         timer := time.NewTimer(getSettings(name).Timeout)
  92.         defer timer.Stop()
  94.         select {
  95.         case <-cmd.finished: // 结束select
  96.             // returnOnce has been executed in another goroutine
  97.         case <-ctx.Done(): // 处理ctx错误
  98.             returnOnce.Do(func() {
  99.                 returnTicket()
  100.                 cmd.errorWithFallback(ctx, ctx.Err())
  101.                 reportAllEvent()
  102.             })
  103.             return
  104.         case <-timer.C: // 处理超时
  105.             returnOnce.Do(func() {
  106.                 returnTicket()
  107.                 cmd.errorWithFallback(ctx, ErrTimeout)
  108.                 reportAllEvent()
  109.             })
  110.             return
  111.         }
  112.     }()
  114.     return cmd.errChan // 错误返回至上层
  115. }

同步的执行方式，如果需要。

  1. func Do(name string, run runFunc, fallback fallbackFunc) error {
  2.     runC := func(ctx context.Context) error {
  3.         return run()
  4.     }
  5.     var fallbackC fallbackFuncC
  6.     if fallback != nil {
  7.         fallbackC = func(ctx context.Context, err error) error {
  8.             return fallback(err)
  9.         }
  10.     }
  11.     return DoC(context.Background(), name, runC, fallbackC) // 具体逻辑如上GoC
  12. }

末

参考：

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注