熔断器hystrix实现

1
2
3
4

hystrix.Go("my_command", func() error {
	// talk to other services
	return nil
}, nil)

1
2
3
4
5
6
7

hystrix.Go("my_command", func() error {
	// talk to other services
	return nil
}, func(err error) error {
	// do this when services are down
	return nil
})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

output := make(chan bool, 1)
errors := hystrix.Go("my_command", func() error {
	// talk to other services
	output <- true
	return nil
}, nil)

select {
case out := <-output:
	// success
case err := <-errors:
	// failure
}

1
2
3
4

err := hystrix.Do("my_command", func() error {
	// talk to other services
	return nil
}, nil)

1
2
3
4
5

hystrix.ConfigureCommand("my_command", hystrix.CommandConfig{
	Timeout:               1000,
	MaxConcurrentRequests: 100,
	ErrorPercentThreshold: 25,
})

1
2
3

hystrixStreamHandler := hystrix.NewStreamHandler()
hystrixStreamHandler.Start()
go http.ListenAndServe(net.JoinHostPort("", "81"), hystrixStreamHandler)

1
2
3
4
5
6
7
8
9

c, err := plugins.InitializeStatsdCollector(&plugins.StatsdCollectorConfig{
	StatsdAddr: "localhost:8125",
	Prefix:     "myapp.hystrix",
})
if err != nil {
	log.Fatalf("could not initialize statsd client: %v", err)
}

metricCollector.Registry.Register(c.NewStatsdCollector)

1
2
3
4
5
6
7
8

var (
    // ErrMaxConcurrency occurs when too many of the same named command are executed at the same time.
    ErrMaxConcurrency = CircuitError{Message: "max concurrency"}
    // ErrCircuitOpen returns when an execution attempt "short circuits". This happens due to the circuit being measured as unhealthy.
    ErrCircuitOpen = CircuitError{Message: "circuit open"}
    // ErrTimeout occurs when the provided function takes too long to execute.
    ErrTimeout = CircuitError{Message: "timeout"}
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

var (
    // DefaultTimeout is how long to wait for command to complete, in milliseconds
    DefaultTimeout = 1000
    // DefaultMaxConcurrent is how many commands of the same type can run at the same time
    DefaultMaxConcurrent = 10
    // DefaultVolumeThreshold is the minimum number of requests needed before a circuit can be tripped due to health
    DefaultVolumeThreshold = 20
    // DefaultSleepWindow is how long, in milliseconds, to wait after a circuit opens before testing for recovery
    DefaultSleepWindow = 5000
    // DefaultErrorPercentThreshold causes circuits to open once the rolling measure of errors exceeds this percent of requests
    DefaultErrorPercentThreshold = 50
    // DefaultLogger is the default logger that will be used in the Hystrix package. By default prints nothing.
    DefaultLogger = NoopLogger{}
)

1

func Configure(cmds map[string]CommandConfig)

1

func ConfigureCommand(name string, config CommandConfig)

1

func Do(name string, run runFunc, fallback fallbackFunc) error

1

func DoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) error

1

func Flush()

1

func GetCircuitSettings() map[string]*Settings

1

func Go(name string, run runFunc, fallback fallbackFunc) chan error

1

func GoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) chan error

1

func SetLogger(l logger)

1
2
3
4

type CircuitBreaker struct {
    Name string
    // contains filtered or unexported fields
}

1

func GetCircuit(name string) (*CircuitBreaker, bool, error)

1

func (circuit *CircuitBreaker) AllowRequest() bool

1

func (circuit *CircuitBreaker) IsOpen() bool

1

func (circuit *CircuitBreaker) ReportEvent(eventTypes []string, start time.Time, runDuration time.Duration) error

1
2
3

type CircuitError struct {
    Message string
}

1

func (e CircuitError) Error() string

1
2
3
4
5
6
7

type CommandConfig struct {
    Timeout                int `json:"timeout"`
    MaxConcurrentRequests  int `json:"max_concurrent_requests"`
    RequestVolumeThreshold int `json:"request_volume_threshold"`
    SleepWindow            int `json:"sleep_window"`
    ErrorPercentThreshold  int `json:"error_percent_threshold"`
}

1

type NoopLogger struct{}

1

func (l NoopLogger) Printf(format string, items ...interface{})

1
2
3
4
5
6
7

type Settings struct {
    Timeout                time.Duration
    MaxConcurrentRequests  int
    RequestVolumeThreshold uint64
    SleepWindow            time.Duration
    ErrorPercentThreshold  int
}

1
2
3

type StreamHandler struct {
    // contains filtered or unexported fields
}

1

func NewStreamHandler() *StreamHandler

1

func (sh *StreamHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request)

1

func (sh *StreamHandler) Start()

1

func (sh *StreamHandler) Stop()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

// Do runs your function in a synchronous manner, blocking until either your function succeeds
// or an error is returned, including hystrix circuit errors
func Do(name string, run runFunc, fallback fallbackFunc) error {
	runC := func(ctx context.Context) error {
		return run()
	}
	var fallbackC fallbackFuncC
	if fallback != nil {
		fallbackC = func(ctx context.Context, err error) error {
			return fallback(err)
		}
	}
	return DoC(context.Background(), name, runC, fallbackC)
}

// DoC runs your function in a synchronous manner, blocking until either your function succeeds
// or an error is returned, including hystrix circuit errors
func DoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) error {
	done := make(chan struct{}, 1)

	r := func(ctx context.Context) error {
		err := run(ctx)
		if err != nil {
			return err
		}

		done <- struct{}{}
		return nil
	}

	f := func(ctx context.Context, e error) error {
		err := fallback(ctx, e)
		if err != nil {
			return err
		}

		done <- struct{}{}
		return nil
	}

	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
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// Go runs your function while tracking the health of previous calls to it.
// If your function begins slowing down or failing repeatedly, we will block
// new calls to it for you to give the dependent service time to repair.
//
// Define a fallback function if you want to define some code to execute during outages.
func Go(name string, run runFunc, fallback fallbackFunc) chan error {
	runC := func(ctx context.Context) error {
		return run()
	}
	var fallbackC fallbackFuncC
	if fallback != nil {
		fallbackC = func(ctx context.Context, err error) error {
			return fallback(err)
		}
	}
	return GoC(context.Background(), name, runC, fallbackC)
}

1
2
3
4
5
6
7
8

// CommandConfig is used to tune circuit settings at runtime
type CommandConfig struct {
	Timeout                int `json:"timeout"`// 超时时间定义
	MaxConcurrentRequests  int `json:"max_concurrent_requests"`// 最大并发请求数
	RequestVolumeThreshold int `json:"request_volume_threshold"`// 跳闸的最小请求数（不健康的断路器）
	SleepWindow            int `json:"sleep_window"`// 跳闸之后可以重试的时间
	ErrorPercentThreshold  int `json:"error_percent_threshold"`// 请求出错比
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// ReportEvent records command metrics for tracking recent error rates and exposing data to the dashboard.
func (circuit *CircuitBreaker) ReportEvent(eventTypes []string, start time.Time, runDuration time.Duration) error {
	if len(eventTypes) == 0 {
		return fmt.Errorf("no event types sent for metrics")
	}

	circuit.mutex.RLock()
	o := circuit.open
	circuit.mutex.RUnlock()
	if eventTypes[0] == "success" && o {
		circuit.setClose()
	}

	var concurrencyInUse float64
	if circuit.executorPool.Max > 0 {
		concurrencyInUse = float64(circuit.executorPool.ActiveCount()) / float64(circuit.executorPool.Max)
	}

	select {
	case circuit.metrics.Updates <- &commandExecution{
		Types:            eventTypes,
		Start:            start,
		RunDuration:      runDuration,
		ConcurrencyInUse: concurrencyInUse,
	}:
	default:
		return CircuitError{Message: fmt.Sprintf("metrics channel (%v) is at capacity", circuit.Name)}
	}

	return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type metricExchange struct {
    Name    string
    Updates chan *commandExecution
    Mutex   *sync.RWMutex

    metricCollectors []metricCollector.MetricCollector
}

type commandExecution struct {
    Types            []string      `json:"types"`
    Start            time.Time     `json:"start_time"`
    RunDuration      time.Duration `json:"run_duration"`
    ConcurrencyInUse float64       `json:"concurrency_inuse"`
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func newMetricExchange(name string) *metricExchange {
    m := &metricExchange{}
    m.Name = name

    m.Updates = make(chan *commandExecution, 2000)
    m.Mutex = &sync.RWMutex{}
    m.metricCollectors = metricCollector.Registry.InitializeMetricCollectors(name)
    m.Reset()

    go m.Monitor()

    return m
}

func (m *metricExchange) Monitor() {
    for update := range m.Updates {
        // we only grab a read lock to make sure Reset() isn't changing the numbers.
        m.Mutex.RLock()

        totalDuration := time.Since(update.Start)
        wg := &sync.WaitGroup{}
        for _, collector := range m.metricCollectors {
            wg.Add(1)
            go m.IncrementMetrics(wg, collector, update, totalDuration)
        }
        wg.Wait()

        m.Mutex.RUnlock()
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func (m *metricExchange) IncrementMetrics(wg *sync.WaitGroup, collector metricCollector.MetricCollector, update *commandExecution, totalDuration time.Duration) {
    // granular metrics
    r := metricCollector.MetricResult{
        Attempts:         1,
        TotalDuration:    totalDuration,
        RunDuration:      update.RunDuration,
        ConcurrencyInUse: update.ConcurrencyInUse,
    }
    switch update.Types[0] {
    case "success":
        r.Successes = 1
    case "failure":
        r.Failures = 1
        r.Errors = 1
    case "rejected":
        r.Rejects = 1
        r.Errors = 1
    // ...
    }
    // ...
    collector.Update(r)
    wg.Done()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// DefaultMetricCollector holds information about the circuit state.
// This implementation of MetricCollector is the canonical source of information about the circuit.
// It is used for for all internal hystrix operations
// including circuit health checks and metrics sent to the hystrix dashboard.
//
// Metric Collectors do not need Mutexes as they are updated by circuits within a locked context.
type DefaultMetricCollector struct {
	mutex *sync.RWMutex

	numRequests *rolling.Number
	errors      *rolling.Number

	successes               *rolling.Number
	failures                *rolling.Number
	rejects                 *rolling.Number
	shortCircuits           *rolling.Number
	timeouts                *rolling.Number
	contextCanceled         *rolling.Number
	contextDeadlineExceeded *rolling.Number

	fallbackSuccesses *rolling.Number
	fallbackFailures  *rolling.Number
	totalDuration     *rolling.Timing
	runDuration       *rolling.Timing
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Number tracks a numberBucket over a bounded number of
// time buckets. Currently the buckets are one second long and only the last 10 seconds are kept.
type Number struct {
	Buckets map[int64]*numberBucket
	Mutex   *sync.RWMutex
}

type numberBucket struct {
    Value float64
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func (r *Number) getCurrentBucket() *numberBucket {
    now := time.Now().Unix()
    var bucket *numberBucket
    var ok bool

    if bucket, ok = r.Buckets[now]; !ok {
        bucket = &numberBucket{}
        r.Buckets[now] = bucket
    }

    return bucket
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

func (r *Number) removeOldBuckets() {
    now := time.Now().Unix() - 10

    for timestamp := range r.Buckets {
        // TODO: configurable rolling window
        if timestamp <= now {
            delete(r.Buckets, timestamp)
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func (r *Number) Increment(i float64) {
    if i == 0 {
        return
    }

    r.Mutex.Lock()
    defer r.Mutex.Unlock()

    b := r.getCurrentBucket()
    b.Value += i
    r.removeOldBuckets()
}

1
2
3
4
5
6

type executorPool struct { // 执行池
    Name    string  // 名字
    Metrics *poolMetrics // 执行池监控
    Max     int // 最大的并发请求数量
    Tickets chan *struct{} // 票证
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

//开启新的执行池
func newExecutorPool(name string) *executorPool {
    p := &executorPool{}
    p.Name = name//名字
    p.Metrics = newPoolMetrics(name)
    p.Max = getSettings(name).MaxConcurrentRequests// 从配置中获取最大的并发请求数量，如果配置中没有，则从默认配置中获取

    p.Tickets = make(chan *struct{}, p.Max)// 初始化buffer chan
    for i := 0; i < p.Max; i++ {
        p.Tickets <- &struct{}{}
    }

    return p
}
//上报状态
func (p *executorPool) Return(ticket *struct{}) {
    if ticket == nil {
        return
    }
    p.Metrics.Updates <- poolMetricsUpdate{
        activeCount: p.ActiveCount(),
    }
    p.Tickets <- ticket
}
func (p *executorPool) ActiveCount() int {
    return p.Max - len(p.Tickets)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

type poolMetrics struct {
	Mutex   *sync.RWMutex
	Updates chan poolMetricsUpdate

	Name              string
	MaxActiveRequests *rolling.Number
	Executed          *rolling.Number
}
type poolMetricsUpdate struct {
	activeCount int
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

// GoC runs your function while tracking the health of previous calls to it.
// If your function begins slowing down or failing repeatedly, we will block
// new calls to it for you to give the dependent service time to repair.
//
// Define a fallback function if you want to define some code to execute during outages.
func GoC(ctx context.Context, name string, run runFuncC, fallback fallbackFuncC) chan error {
	cmd := &command{// command执行者
		run:      run, //run
		fallback: fallback,//fallback
		start:    time.Now(), //开始时间
		errChan:  make(chan error, 1), //错误
		finished: make(chan bool, 1), //是否完成
	}

	// dont have methods with explicit params and returns
	// let data come in and out naturally, like with any closure
	// explicit error return to give place for us to kill switch the operation (fallback)
    //得到断路器,不存在则创建
	circuit, _, err := GetCircuit(name)
	if err != nil {
		cmd.errChan <- err
		return cmd.errChan
	}
	cmd.circuit = circuit
	ticketCond := sync.NewCond(cmd)// cond条件
	ticketChecked := false
	// When the caller extracts error from returned errChan, it's assumed that
	// the ticket's been returned to executorPool. Therefore, returnTicket() can
    // not run after cmd.errorWithFallback().
    // 返还ticket
	returnTicket := func() {
		cmd.Lock()
		// Avoid releasing before a ticket is acquired.
		for !ticketChecked {
			ticketCond.Wait()// 相当于select{}
		}
		cmd.circuit.executorPool.Return(cmd.ticket)// 将ticket放回池子中
		cmd.Unlock()
	}
	// Shared by the following two goroutines. It ensures only the faster
    // goroutine runs errWithFallback() and reportAllEvent().
    // 上报执行状态
	returnOnce := &sync.Once{}// 确保被multi goroutine执行一次
	reportAllEvent := func() {// events采集，后续dashboard使用
		err := cmd.circuit.ReportEvent(cmd.events, cmd.start, cmd.runDuration)
		if err != nil {
			log.Printf(err.Error())
		}
	}
    // g1, 检测断路器不允许通过，尝试fallback，将中途遇到的event上报。
	go func() {
		defer func() { cmd.finished <- true }()

		// Circuits get opened when recent executions have shown to have a high error rate.
		// Rejecting new executions allows backends to recover, and the circuit will allow
        // new traffic when it feels a healthly state has returned.
        // 查看断路器是否已打开
		if !cmd.circuit.AllowRequest() {
			cmd.Lock()
			// It's safe for another goroutine to go ahead releasing a nil ticket.
			ticketChecked = true
			ticketCond.Signal()
			cmd.Unlock()
			returnOnce.Do(func() {
				returnTicket()
				cmd.errorWithFallback(ctx, ErrCircuitOpen)
				reportAllEvent()
			})
			return
		}

		// As backends falter, requests take longer but don't always fail.
		//
		// When requests slow down but the incoming rate of requests stays the same, you have to
		// run more at a time to keep up. By controlling concurrency during these situations, you can
		// shed load which accumulates due to the increasing ratio of active commands to incoming requests.
        cmd.Lock()
        // 获取ticket 如果得不到就限流
		select {
		case cmd.ticket = <-circuit.executorPool.Tickets:// 从池子中取出ticket
			ticketChecked = true
			ticketCond.Signal()// 通知cond.Wait()
			cmd.Unlock()
		default:
			ticketChecked = true
			ticketCond.Signal()
			cmd.Unlock()
			returnOnce.Do(func() {
				returnTicket()
				cmd.errorWithFallback(ctx, ErrMaxConcurrency)// 并发过高
				reportAllEvent()
			})
			return
		}
        // 执行我们自已的方法，并上报执行信息
		runStart := time.Now()
        runErr := run(ctx)
		returnOnce.Do(func() {
			defer reportAllEvent()
			cmd.runDuration = time.Since(runStart)// 运行时间
			returnTicket()// 把ticket返回去
			if runErr != nil {
				cmd.errorWithFallback(ctx, runErr)
				return
			}
			cmd.reportEvent("success")// 执行成功
		})
	}()
    // 等待context是否被结束，或执行者超时，并上报
	go func() {
		timer := time.NewTimer(getSettings(name).Timeout)
		defer timer.Stop()

		select {
		case <-cmd.finished:// 结束select
			// returnOnce has been executed in another goroutine
		case <-ctx.Done():// 处理ctx错误
			returnOnce.Do(func() {
				returnTicket()
				cmd.errorWithFallback(ctx, ctx.Err())
				reportAllEvent()
			})
			return
		case <-timer.C:// 处理超时
			returnOnce.Do(func() {
				returnTicket()
				cmd.errorWithFallback(ctx, ErrTimeout)
				reportAllEvent()
			})
			return
		}
	}()

	return cmd.errChan// 错误返回至上层
}

1
2
3

hystrixStreamHandler := hystrix.NewStreamHandler()
hystrixStreamHandler.Start()
go http.ListenAndServe(net.JoinHostPort("", "81"), hystrixStreamHandler)