go-pattern-examples/resiliency/06_circuit_breaker/circuit_breaker.go

175 lines
4.0 KiB
Go
Raw Normal View History

2020-05-08 10:51:33 +03:00
package circuit
2020-05-10 17:03:24 +03:00
/*
* @Description: https://github.com/crazybber
* @Author: Edward
* @Date: 2020-05-10 22:00:58
* @Last Modified by: Edward
2020-05-11 17:15:58 +03:00
* @Last Modified time: 2020-05-11 22:15:25
2020-05-10 17:03:24 +03:00
*/
2020-05-08 10:51:33 +03:00
import (
"context"
"errors"
2020-05-10 17:03:24 +03:00
"sync"
2020-05-08 10:51:33 +03:00
"time"
)
2020-05-10 17:03:24 +03:00
////////////////////////////////
///使用HTTP请求的例子
//每个搜索引擎时时刻刻都会遇到超大规模的请求的流量.
//这里演示一个复杂一点的例子同时使用Option 模式
2020-05-08 10:51:33 +03:00
//ErrServiceUnavailable for error
var (
2020-05-08 14:34:48 +03:00
ErrTooManyRequests = errors.New("too many requests")
ErrServiceUnavailable = errors.New("service unavailable")
2020-05-08 10:51:33 +03:00
FailureThreshold = 10
)
2020-05-10 17:03:24 +03:00
//StateCheckerHandler check state
type StateCheckerHandler func(counts counters) bool
//StateChangedEventHandler set event handle
type StateChangedEventHandler func(name string, from State, to State)
//Option set Options
type Option func(opts *Options)
//RequestBreaker for protection
type RequestBreaker struct {
options Options
mutex sync.Mutex
state State
generation uint64
2020-05-11 13:01:25 +03:00
counts ICounter
2020-05-10 17:03:24 +03:00
}
//NewRequestBreaker return a breaker
2020-05-11 09:31:25 +03:00
func NewRequestBreaker(opts ...Option) *RequestBreaker {
defaultOptions := Options{
Name: "defaultBreakerName",
Expiry: time.Now().Add(time.Second * 20),
Interval: time.Second * 2,
2020-05-11 13:01:25 +03:00
Timeout: time.Second * 60, //default to 60 seconds
2020-05-11 09:31:25 +03:00
MaxRequests: 5,
ReadyToTrip: func(counts counters) bool { return true },
OnStateChanged: func(name string, from State, to State) {},
}
for _, setOption := range opts {
setOption(&defaultOptions)
}
return &RequestBreaker{
options: defaultOptions,
counts: nil,
generation: 0,
}
2020-05-10 17:03:24 +03:00
}
2020-05-15 13:31:56 +03:00
// Do the given requested work if the RequestBreaker accepts it.
2020-05-14 18:35:44 +03:00
// Do returns an error instantly if the RequestBreaker rejects the request.
// Otherwise, Execute returns the result of the request.
// If a panic occurs in the request, the RequestBreaker handles it as an error and causes the same panic again.
2020-05-15 13:31:56 +03:00
func (rb *RequestBreaker) Do(work func() (interface{}, error)) (interface{}, error) {
//do work from requested user
result, err := work()
2020-05-14 18:35:44 +03:00
return result, err
}
2020-05-08 10:51:33 +03:00
//State of current switch
2020-05-08 14:34:48 +03:00
type State int
//states of CircuitBreaker
2020-05-08 10:51:33 +03:00
const (
2020-05-08 14:34:48 +03:00
UnknownState State = iota
2020-05-08 10:51:33 +03:00
FailureState
SuccessState
)
//Circuit of action stream
type Circuit func(context.Context) error
2020-05-11 13:01:25 +03:00
//ICounter interface
type ICounter interface {
2020-05-08 14:34:48 +03:00
Count(State)
2020-05-08 10:51:33 +03:00
LastActivity() time.Time
Reset()
}
type counters struct {
2020-05-11 17:15:58 +03:00
Requests uint32
lastState State
lastActivity time.Time
counts uint32 //counts of failures
TotalFailures uint32
TotalSuccesses uint32
ConsecutiveSuccesses uint32
ConsecutiveFailures uint32
2020-05-08 10:51:33 +03:00
}
func (c *counters) LastActivity() time.Time {
return c.lastActivity
}
func (c *counters) Reset() {
}
2020-05-15 13:31:56 +03:00
//Count the failure and success
2020-05-11 17:15:58 +03:00
func (c *counters) Count(statue State) {
switch statue {
case FailureState:
c.ConsecutiveFailures++
case SuccessState:
c.ConsecutiveSuccesses++
}
c.Requests++
c.lastState = statue
2020-05-08 10:51:33 +03:00
}
2020-05-15 13:31:56 +03:00
//WrapperBreaker return a Wrapper to hold request
func WrapperBreaker(c Circuit, failureThreshold uint32) Circuit {
2020-05-08 10:51:33 +03:00
2020-05-15 13:31:56 +03:00
//内部计数器
2020-05-11 17:15:58 +03:00
cnt := counters{}
2020-05-08 10:51:33 +03:00
return func(ctx context.Context) error {
2020-05-11 17:15:58 +03:00
if cnt.ConsecutiveFailures >= failureThreshold {
canRetry := func(cnt counters) bool {
2020-05-11 17:00:45 +03:00
2020-05-11 17:15:58 +03:00
backoffLevel := cnt.ConsecutiveFailures - failureThreshold
2020-05-08 10:51:33 +03:00
// Calculates when should the circuit breaker resume propagating requests
// to the service
shouldRetryAt := cnt.LastActivity().Add(time.Second * 2 << backoffLevel)
return time.Now().After(shouldRetryAt)
}
if !canRetry(cnt) {
// Fails fast instead of propagating requests to the circuit since
// not enough time has passed since the last failure to retry
return ErrServiceUnavailable
}
}
// Unless the failure threshold is exceeded the wrapped service mimics the
// old behavior and the difference in behavior is seen after consecutive failures
if err := c(ctx); err != nil {
cnt.Count(FailureState)
return err
}
cnt.Count(SuccessState)
return nil
}
}