retry

merger/merge.go

@@ -33,6 +33,7 @@ func (f *FanIn) Start() {
 	go f.start(st)
 	<-st
 }
+
 func (f *FanIn) start(st chan struct{}) {
 	st <- struct{}{}
 	for {

utils/xretry/xretry.go

+package xretry
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"math/rand"
+	"reflect"
+	"runtime"
+	"strings"
+	"time"
+)
+
+const (
+	DefaultRetryTimes          = 5
+	DefaultRetryLinearInterval = time.Second * 3
+)
+
+type RetryConfig struct {
+	context         context.Context
+	retryTimes      uint
+	backoffStrategy BackoffStrategy
+}
+
+type RetryFunc func() error
+
+type Option func(*RetryConfig)
+
+func RetryTimes(n uint) Option {
+	return func(rc *RetryConfig) {
+		rc.retryTimes = n
+	}
+}
+
+func RetryWithCustomBackoff(backoffStrategy BackoffStrategy) Option {
+	if backoffStrategy == nil {
+		panic("backoffStrategy不能为空")
+	}
+
+	return func(rc *RetryConfig) {
+		rc.backoffStrategy = backoffStrategy
+	}
+}
+
+func RetryWithLinearBackoff(interval time.Duration) Option {
+	if interval <= 0 {
+		panic("retry间隔必须大于0")
+	}
+
+	return func(rc *RetryConfig) {
+		rc.backoffStrategy = &linear{
+			interval: interval,
+		}
+	}
+}
+
+func RetryWithExponentialWithJitterBackoff(interval time.Duration, base uint64, maxJitter time.Duration) Option {
+	if interval <= 0 {
+		panic("retry间隔必须大于0")
+	}
+
+	if maxJitter < 0 {
+		panic("maxJitter必须大于0")
+	}
+
+	if base%2 == 0 {
+		return func(rc *RetryConfig) {
+			rc.backoffStrategy = &shiftExponentialWithJitter{
+				interval:  interval,
+				maxJitter: maxJitter,
+				shifter:   uint64(math.Log2(float64(base))),
+			}
+		}
+	}
+
+	return func(rc *RetryConfig) {
+		rc.backoffStrategy = &exponentialWithJitter{
+			interval:  interval,
+			base:      time.Duration(base),
+			maxJitter: maxJitter,
+		}
+	}
+}
+
+func Context(ctx context.Context) Option {
+	return func(rc *RetryConfig) {
+		rc.context = ctx
+	}
+}
+
+func Retry(retryFunc RetryFunc, opts ...Option) error {
+	config := &RetryConfig{
+		retryTimes: DefaultRetryTimes,
+		context:    context.TODO(),
+	}
+
+	for _, opt := range opts {
+		opt(config)
+	}
+
+	if config.backoffStrategy == nil {
+		config.backoffStrategy = &linear{
+			interval: DefaultRetryLinearInterval,
+		}
+	}
+
+	var i uint
+	for i < config.retryTimes {
+		err := retryFunc()
+		if err != nil {
+			select {
+			case <-time.After(config.backoffStrategy.CalculateInterval()):
+			case <-config.context.Done():
+				return errors.New("retry is cancelled")
+			}
+		} else {
+			return nil
+		}
+		i++
+	}
+
+	funcPath := runtime.FuncForPC(reflect.ValueOf(retryFunc).Pointer()).Name()
+	lastSlash := strings.LastIndex(funcPath, "/")
+	funcName := funcPath[lastSlash+1:]
+
+	return fmt.Errorf("function %s run failed after %d times retry", funcName, i)
+}
+
+type BackoffStrategy interface {
+	CalculateInterval() time.Duration
+}
+
+type linear struct {
+	interval time.Duration
+}
+
+func (l linear) CalculateInterval() time.Duration {
+	return l.interval
+}
+
+type exponentialWithJitter struct {
+	base      time.Duration
+	interval  time.Duration
+	maxJitter time.Duration
+}
+
+func (e *exponentialWithJitter) CalculateInterval() time.Duration {
+	current := e.interval
+	e.interval = e.interval * e.base
+	return current + jitter(e.maxJitter)
+}
+
+type shiftExponentialWithJitter struct {
+	interval  time.Duration
+	maxJitter time.Duration
+	shifter   uint64
+}
+
+func (e shiftExponentialWithJitter) CalculateInterval() time.Duration {
+	current := e.interval
+	e.interval = e.interval << e.shifter
+	return current + jitter(e.maxJitter)
+}
+
+func jitter(maxJitter time.Duration) time.Duration {
+	if maxJitter == 0 {
+		return 0
+	}
+	return time.Duration(rand.Int63n(int64(maxJitter)) + 1)
+}