/* * @Description: https://github.com/crazybber * @Author: Edward * @Date: 2020-05-10 22:00:58 * @Last Modified by: Edward * @Last Modified time: 2020-06-03 23:54:29 */ package circuit import ( "context" "errors" "fmt" "sync" "time" ) //////////////////////////////// ///使用HTTP请求的例子 //每个搜索引擎时时刻刻都会遇到超大规模的请求的流量. //这里演示一个复杂一点的例子,同时使用Option 模式 //ErrServiceUnavailable for error var ( ErrTooManyRequests = errors.New("too many requests") ErrServiceUnavailable = errors.New("service unavailable") FailureThreshold = 10 //最大失败次数--->失败阈值 ) // 默认的超时时间 const ( defaultTimeout = time.Second * 30 defaultSuccessThreshold = 2 ) //////////////////////////////// //way 2 对象式断路器 // 高级模式 // 支持多工作者模式 //////////////////////////////// //RequestBreaker for protection type RequestBreaker struct { options Options mutex sync.Mutex state State cnter counters preState State } //NewRequestBreaker return a breaker func NewRequestBreaker(opts ...Option) *RequestBreaker { defaultOptions := Options{ Name: "defaultBreakerName", Expiry: time.Now().Add(time.Second * 20), Interval: time.Second * 10, // interval to check closed status,default 10 seconds Timeout: time.Second * 60, //timeout to check open, default 60 seconds MaxRequests: 5, CanOpen: func(current State, cnter counters) bool { return cnter.ConsecutiveFailures > 2 }, CanClose: func(current State, cnter counters) bool { return cnter.ConsecutiveSuccesses > 2 }, OnStateChanged: func(name string, fromPre State, toCurrent State) {}, } for _, setOption := range opts { setOption(&defaultOptions) } return &RequestBreaker{ options: defaultOptions, cnter: counters{}, state: StateUnknown, preState: StateUnknown, } } func (rb *RequestBreaker) changeStateTo(state State) { rb.preState = rb.state rb.state = state rb.cnter.Reset() } func (rb *RequestBreaker) beforeRequest() error { rb.mutex.Lock() defer rb.mutex.Unlock() fmt.Println("before do request:", rb.cnter.Total()) switch rb.state { case StateOpen: //如果是断开状态,并且超时了,转到半开状态,记录 if rb.options.Expiry.Before(time.Now()) { rb.changeStateTo(StateHalfOpen) rb.options.Expiry = time.Now().Add(rb.options.Timeout) return nil } return ErrTooManyRequests case StateClosed: if rb.options.Expiry.Before(time.Now()) { rb.cnter.Reset() rb.options.Expiry = time.Now().Add(rb.options.Interval) } } return nil } // Do the given requested work if the RequestBreaker accepts it. // 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. func (rb *RequestBreaker) Do(work func(ctx context.Context) (interface{}, error)) (interface{}, error) { //before if err := rb.beforeRequest(); err != nil { return nil, err } //do work //do work from requested user result, err := work(rb.options.Ctx) //after work rb.afterRequest(err) return result, err } func (rb *RequestBreaker) afterRequest(resultErr error) { rb.mutex.Lock() defer rb.mutex.Unlock() //after fmt.Println("after do request:", rb.cnter.Total()) if resultErr != nil { //失败了,handle 失败 rb.cnter.Count(FailureState, rb.preState == rb.state) switch rb.state { case StateHalfOpen, StateClosed: if rb.options.CanOpen(rb.state, rb.cnter) { rb.changeStateTo(StateOpen) //打开开关 rb.options.OnStateChanged(rb.options.Name, rb.state, StateOpen) //关闭到打开 } } } else { //success ! rb.cnter.Count(SuccessState, rb.preState == rb.state) switch rb.state { case StateHalfOpen: if rb.preState == StateOpen { // rb.changeStateTo(StateHalfOpen) //already handled in beforeRequest,Only fire StateChange Event rb.options.OnStateChanged(rb.options.Name, StateOpen, StateHalfOpen) //打开到半开 } if rb.cnter.ConsecutiveSuccesses >= rb.options.ShoulderHalfToOpen { rb.changeStateTo(StateClosed) rb.options.Expiry = time.Now().Add(rb.options.Interval) rb.options.OnStateChanged(rb.options.Name, StateHalfOpen, StateClosed) //半开到关闭 } } } }