Kratos 源码分析：熔断

1
2
3
4
5

type CircuitBreaker interface {
	Allow() error
	MarkSuccess()
	MarkFailed()
}

1
2
3
4
5
6

type Group struct {
   mutex sync.Mutex
   val   atomic.Value
 
   New func() CircuitBreaker
}

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

// Get .
func (g *Group) Get(name string) CircuitBreaker {
	m, ok := g.val.Load().(map[string]CircuitBreaker)
	if ok {
		breaker, ok := m[name]
		if ok {
			return breaker // 很具name从val拿出 breaker 如果存在返回
		}
	}
	// slowpath for group don`t have specified name breaker.
	g.mutex.Lock()
	nm := make(map[string]CircuitBreaker, len(m)+1)
	for k, v := range m {
		nm[k] = v
	}
	breaker := g.New()
	nm[name] = breaker // 如果不存在 生成一个 并放入map 并返回
	g.val.Store(nm)
	g.mutex.Unlock()
	return breaker
}

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

// Breaker is a sre CircuitBreaker pattern.
type Breaker struct {
   stat window.RollingCounter
   r    *rand.Rand
   // rand.New(...) returns a non thread safe object
   randLock sync.Mutex
 
   // Reducing the k will make adaptive throttling behave more aggressively,
   // Increasing the k will make adaptive throttling behave less aggressively.
   k       float64
   request int64
 
   state int32
}

 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

// Allow request if error returns nil.
func (b *Breaker) Allow() error {
	success, total := b.summary() // 从活动窗口获取成功数和总数
	k := b.k * float64(success) // 根据k成功系数 获取
 
	// check overflow requests = K * success
	if total < b.request || float64(total) < k { // 如果总数<request 或者  总数 < k 
		if atomic.LoadInt32(&b.state) == StateOpen { 
			atomic.CompareAndSwapInt32(&b.state, StateOpen, StateClosed) // 如果state是打开 关闭
		}
		return nil
	}
	if atomic.LoadInt32(&b.state) == StateClosed { 
		atomic.CompareAndSwapInt32(&b.state, StateClosed, StateOpen) // 如果state是关闭 打开
	}
	dr := math.Max(0, (float64(total)-k)/float64(total+1)) // 获取系数，当k越大 dr越小
	drop := b.trueOnProba(dr)// trueOnProba 获取水机数，当dr越大，drop越大
	if drop { // 如果是 拒绝请求
		return circuitbreaker.ErrNotAllowed
	}
	return nil
}
 
func (b *Breaker) trueOnProba(proba float64) (truth bool) {
	b.randLock.Lock()
	truth = b.r.Float64() < proba
	b.randLock.Unlock()
	return
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// MarkSuccess mark requeest is success.
func (b *Breaker) MarkSuccess() {
	b.stat.Add(1) // 成功数+1
}
 
// MarkFailed mark request is failed.
func (b *Breaker) MarkFailed() {
	// NOTE: when client reject requets locally, continue add counter let the
	// drop ratio higher.
	b.stat.Add(0) // 失败数+1
}