Golang中Retry重試實踐
使用github.com/kamilsk/retry/v5包,核心API:
// Do takes the action and performs it, repetitively, until successful.
//
// Optionally, strategies may be passed that assess whether or not an attempt
// should be made.
func Do(
breaker Breaker,
action func(context.Context) error,
strategies ...func(Breaker, uint, error) bool,
) error {
var (
ctx = convert(breaker)
err error = internal
core error
)
for attempt, should := uint(0), true; should; attempt++ {
core = unwrap(err)
for i, repeat := 0, len(strategies); should && i < repeat; i++ {
should = should && strategies[i](breaker, attempt, core)
}
select {
case <-breaker.Done():// 若有中斷信號,則直接返回
return breaker.Err()
default:// 沒有中斷信號,則根據(jù)策略決定是否執(zhí)行
if should {
err = action(ctx)
}
}
should = should && err != nil
}
return err
}
// Action defines a callable function that package retry can handle.
type Action = func(context.Context) error
// 中斷器帶有「中斷信號」
// A Breaker carries a cancellation signal to interrupt an action execution.
//
// It is a subset of the built-in context and github.com/kamilsk/breaker interfaces.
type Breaker = interface {
// Done returns a channel that's closed when a cancellation signal occurred.
Done() <-chan struct{}
// If Done is not yet closed, Err returns nil.
// If Done is closed, Err returns a non-nil error.
// After Err returns a non-nil error, successive calls to Err return the same error.
Err() error
}
由兩個維度控制執(zhí)行行為:一個是「Breaker中斷器」,一個是「strategies策略」。
中斷器
具體有哪些中斷器見github.com/kamilsk/breaker(也可自定義中斷器):
// Interface carries a cancellation signal to interrupt an action execution.
//
// Example based on github.com/kamilsk/retry/v5 module:
//
// if err := retry.Do(breaker.BreakByTimeout(time.Minute), action); err != nil {
// log.Fatal(err)
// }
//
// Example based on github.com/kamilsk/semaphore/v5 module:
//
// if err := semaphore.Acquire(breaker.BreakByTimeout(time.Minute), 5); err != nil {
// log.Fatal(err)
// }
//
type Interface interface {
// Close closes the Done channel and releases resources associated with it.
Close()
// Done returns a channel that's closed when a cancellation signal occurred.
Done() <-chan struct{}
// If Done is not yet closed, Err returns nil.
// If Done is closed, Err returns a non-nil error.
// After Err returns a non-nil error, successive calls to Err return the same error.
Err() error
// trigger is a private method to guarantee that the breakers come from
// this package and all of them return a valid Done channel.
trigger() Interface
}
// 基于信號的中斷器
func BreakByChannel(signal <-chan struct{}) Interface {
return (&channelBreaker{newBreaker(), signal}).trigger()
}
type channelBreaker struct {
*breaker
relay <-chan struct{}
}
// Close closes the Done channel and releases resources associated with it.
func (br *channelBreaker) Close() {
br.closer.Do(func() {
close(br.signal)
})
}
// trigger starts listening to the internal signal to close the Done channel.
func (br *channelBreaker) trigger() Interface {
go func() {
select {
case <-br.relay:
case <-br.signal:
}
br.Close()
// the goroutine is done
atomic.StoreInt32(&br.released, 1)
}()
return br
}
// 基于超時時間的中斷器
func BreakByTimeout(timeout time.Duration) Interface {
if timeout < 0 {
return closedBreaker()
}
return newTimeoutBreaker(timeout).trigger()
}
type timeoutBreaker struct {
*breaker
*time.Timer
}
// Close closes the Done channel and releases resources associated with it.
func (br *timeoutBreaker) Close() {
br.closer.Do(func() {
stop(br.Timer)
close(br.signal)
})
}
// trigger starts listening to the internal timer to close the Done channel.
func (br *timeoutBreaker) trigger() Interface {
go func() {
select {
case <-br.Timer.C:
case <-br.signal:
}
br.Close()
// the goroutine is done
atomic.StoreInt32(&br.released, 1)
}()
return br
}
func stop(timer *time.Timer) {
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
}
func newBreaker() *breaker {
return &breaker{signal: make(chan struct{})}
}
type breaker struct {
closer sync.Once
signal chan struct{}
released int32
}
// Close closes the Done channel and releases resources associated with it.
func (br *breaker) Close() {
br.closer.Do(func() {
close(br.signal)
atomic.StoreInt32(&br.released, 1)
})
}
// Done returns a channel that's closed when a cancellation signal occurred.
func (br *breaker) Done() <-chan struct{} {
return br.signal
}
// Err returns a non-nil error if the Done channel is closed and nil otherwise.
// After Err returns a non-nil error, successive calls to Err return the same error.
func (br *breaker) Err() error {
if atomic.LoadInt32(&br.released) == 1 {
return Interrupted
}
return nil
}
// IsReleased returns true if resources associated with the breaker were released.
func (br *breaker) IsReleased() bool {
return atomic.LoadInt32(&br.released) == 1
}
func (br *breaker) trigger() Interface {
return br
}
組合多個中斷器:
func Multiplex(breakers ...Interface) Interface {
if len(breakers) == 0 {
return closedBreaker()
}
for len(breakers) < 3 {
breakers = append(breakers, stub{})
}
return newMultiplexedBreaker(breakers).trigger()
}
func newMultiplexedBreaker(breakers []Interface) *multiplexedBreaker {
return &multiplexedBreaker{newBreaker(), breakers}
}
type multiplexedBreaker struct {
*breaker
breakers []Interface
}
// Close closes the Done channel and releases resources associated with it.
func (br *multiplexedBreaker) Close() {
br.closer.Do(func() {
each(br.breakers).Close()
close(br.signal)
})
}
// trigger starts listening to the all Done channels of multiplexed breakers.
func (br *multiplexedBreaker) trigger() Interface {
go func() {
if len(br.breakers) == 3 {
select {
case <-br.breakers[0].Done():
case <-br.breakers[1].Done():
case <-br.breakers[2].Done():
}
} else {
brs := make([]reflect.SelectCase, 0, len(br.breakers))
for _, br := range br.breakers {
brs = append(brs, reflect.SelectCase{
Dir: reflect.SelectRecv,
Chan: reflect.ValueOf(br.Done()),
})
}
reflect.Select(brs)
}
br.Close()
// the goroutine is done
atomic.StoreInt32(&br.released, 1)
}()
return br
}
策略
具體策略可見github.com/kamilsk/retry/v5/strategy包(也可自定義策略):
type Strategy = func(breaker Breaker, attempt uint, err error) bool
// 重試次數(shù)
// Limit creates a Strategy that limits the number of attempts
// that Retry will make.
func Limit(value uint) Strategy {
return func(_ Breaker, attempt uint, _ error) bool {
return attempt < value
}
}
// 延遲執(zhí)行
// Delay creates a Strategy that waits the given duration
// before the first attempt is made.
func Delay(duration time.Duration) Strategy {
return func(breaker Breaker, attempt uint, _ error) bool {
keep := true
if attempt == 0 {
timer := time.NewTimer(duration)
select {
case <-timer.C:
case <-breaker.Done():
keep = false
}
stop(timer)
}
return keep
}
}
// 延遲退避
// Backoff creates a Strategy that waits before each attempt, with a duration as
// defined by the given backoff.Algorithm.
func Backoff(algorithm func(attempt uint) time.Duration) Strategy {
return BackoffWithJitter(algorithm, func(duration time.Duration) time.Duration {
return duration
})
}
// BackoffWithJitter creates a Strategy that waits before each attempt, with a
// duration as defined by the given backoff.Algorithm and jitter.Transformation.
func BackoffWithJitter(
algorithm func(attempt uint) time.Duration,
transformation func(duration time.Duration) time.Duration,
) Strategy {
return func(breaker Breaker, attempt uint, _ error) bool {
keep := true
if attempt > 0 {
timer := time.NewTimer(transformation(algorithm(attempt)))
select {
case <-timer.C:
case <-breaker.Done():
keep = false
}
stop(timer)
}
return keep
}
}
到此這篇關(guān)于Golang中Retry重試實踐的文章就介紹到這了,更多相關(guān)Golang Retry重試內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
go mod 安裝依賴 unkown revision問題的解決方案
這篇文章主要介紹了go mod 安裝依賴 unkown revision問題的解決方案,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2021-05-05
Golang中實現(xiàn)數(shù)據(jù)脫敏處理的go-mask包分享
這篇文章主要是來和大家分享一個在輸出中對敏感數(shù)據(jù)進行脫敏的工作包:go-mask,可以將敏感信息輸出的時候替換成星號或其他字符,感興趣的小編可以跟隨小編一起了解下2023-05-05
關(guān)于Golang變量初始化/類型推斷/短聲明的問題
這篇文章主要介紹了關(guān)于Golang變量初始化/類型推斷/短聲明的問題,本文給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-02-02
Go語言操作金倉數(shù)據(jù)庫之SQL執(zhí)行,類型映射與超時控制詳解
本文詳細介紹了使用Go語言操作金倉數(shù)據(jù)庫的關(guān)鍵技術(shù)點,主要包括 SQL執(zhí)行,通過Query方法執(zhí)行查詢并處理結(jié)果集,使用Exec執(zhí)行非查詢操作,以及事務(wù)處理多條SQL語句的方法,有需要的小伙伴可以了解下2026-05-05

