Golang中Retry重试实践

// 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:// 没有中断信号，则根据策略决定是否执行
          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
}

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

type Strategy = func(breaker Breaker, attempt uint, err error) bool

// 重试次数
// 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
    }
}

// 延迟执行
// 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
    }
}