《游戏服务端编程实践》14.2 重试、断路器与回退机制

// file: ch14/breaker/breaker.go
// tier: L1
// verify: go test ./ch14/breaker/

package breaker

import (
	"context"
	"errors"
	"math/rand"
	"sync"
	"time"
)

// ErrOpen 表示断路器已打开，请求被快速失败。
var ErrOpen = errors.New("breaker: circuit open")

// State 是断路器的三态。
type State uint8

const (
	Closed   State = iota // 正常放行并统计结果
	Open                  // 拒绝请求，等待冷却
	HalfOpen              // 只放行少量探测
)

// Config 描述断路器参数：Closed 保留最近 WindowSize 次结果，
// 样本数达到 MinSamples 且失败率不低于 FailureRate 时打开；
// 冷却 OpenTimeout 后最多并发 HalfOpenProbes 个探测，连续成功
// HalfOpenSuccesses 次才恢复 Closed。
type Config struct {
	WindowSize        int
	MinSamples        int
	FailureRate       float64
	OpenTimeout       time.Duration
	HalfOpenProbes    int
	HalfOpenSuccesses int
}

func (c Config) validate() error {
	switch {
	case c.WindowSize < 1 || c.MinSamples < 1 || c.MinSamples > c.WindowSize:
		return errors.New("breaker: invalid window or samples")
	case c.FailureRate <= 0 || c.FailureRate > 1:
		return errors.New("breaker: failure rate must be in (0,1]")
	case c.OpenTimeout <= 0:
		return errors.New("breaker: open timeout must be positive")
	case c.HalfOpenProbes < 1 || c.HalfOpenSuccesses < 1:
		return errors.New("breaker: invalid half-open probes")
	}
	return nil
}

// Breaker 是并发安全的断路器。时间由 now 注入，测试可以推进
// 虚拟时钟；Allow 成功后必须调用且只调用一次 Done。
type Breaker struct {
	mu            sync.Mutex
	cfg           Config
	state         State
	results       []bool
	next          int
	samples       int
	failures      int
	openedAt      time.Time
	probes        int
	halfSuccesses int
	now           func() time.Time
}

// New 校验配置并创建处于 Closed 状态的断路器。
func New(cfg Config) (*Breaker, error) {
	if err := cfg.validate(); err != nil {
		return nil, err
	}
	return &Breaker{cfg: cfg, state: Closed,
		results: make([]bool, cfg.WindowSize), now: time.Now}, nil
}

// Allow 判断请求能否放行，并在 HalfOpen 占用一个探测名额。
func (b *Breaker) Allow() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.state == Open {
		if b.now().Sub(b.openedAt) < b.cfg.OpenTimeout {
			return ErrOpen
		}
		b.state = HalfOpen
		b.probes, b.halfSuccesses = 0, 0
	}
	if b.state == HalfOpen {
		if b.probes >= b.cfg.HalfOpenProbes {
			return ErrOpen
		}
		b.probes++
	}
	return nil
}

// Done 上报一次已放行请求的结果。Open 状态下的迟到结果被忽略，
// 它们属于上一轮探测，不能污染新一轮半开统计。
func (b *Breaker) Done(success bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	switch b.state {
	case Closed:
		b.recordLocked(success)
	case HalfOpen:
		if b.probes > 0 {
			b.probes--
		}
		if !success {
			b.openLocked()
			return
		}
		b.halfSuccesses++
		if b.halfSuccesses >= b.cfg.HalfOpenSuccesses {
			b.closeLocked()
		}
	}
}

// State 返回当前状态。
func (b *Breaker) State() State { b.mu.Lock(); defer b.mu.Unlock(); return b.state }
func (b *Breaker) recordLocked(success bool) {
	if b.samples == len(b.results) {
		if !b.results[b.next] {
			b.failures--
		}
	} else {
		b.samples++
	}
	b.results[b.next] = success
	if !success {
		b.failures++
	}
	b.next = (b.next + 1) % len(b.results)
	rate := float64(b.failures) / float64(b.samples)
	if b.samples >= b.cfg.MinSamples && rate >= b.cfg.FailureRate {
		b.openLocked()
	}
}
func (b *Breaker) openLocked() {
	b.state = Open
	b.openedAt = b.now()
	b.probes, b.halfSuccesses = 0, 0
}
func (b *Breaker) closeLocked() {
	b.state = Closed
	b.next, b.samples, b.failures = 0, 0, 0
	b.probes, b.halfSuccesses = 0, 0
	for i := range b.results {
		b.results[i] = false
	}
}

// RetryConfig 描述指数退避重试参数。Retryable 必须显式给出，
// 防止把参数错误、权限不足等永久失败也拿去重试。
type RetryConfig struct {
	MaxAttempts int
	BaseDelay   time.Duration
	MaxDelay    time.Duration
	Jitter      float64
	Retryable   func(error) bool
}

func (c RetryConfig) validate() error {
	switch {
	case c.MaxAttempts < 1:
		return errors.New("breaker: max attempts must be at least 1")
	case c.BaseDelay <= 0:
		return errors.New("breaker: base delay must be positive")
	case c.MaxDelay < c.BaseDelay:
		return errors.New("breaker: max delay must not be smaller than base delay")
	case c.Jitter < 0 || c.Jitter > 1:
		return errors.New("breaker: jitter must be in [0,1]")
	case c.Retryable == nil:
		return errors.New("breaker: retryable predicate is required")
	}
	return nil
}

// Retrier 按配置执行带抖动的指数退避重试。
type Retrier struct {
	cfg   RetryConfig
	rnd   *rand.Rand
	sleep func(context.Context, time.Duration) error
}

// NewRetrier 校验配置并创建一个重试器。
func NewRetrier(cfg RetryConfig) (*Retrier, error) {
	if err := cfg.validate(); err != nil {
		return nil, err
	}
	return &Retrier{cfg: cfg,
		rnd:   rand.New(rand.NewSource(time.Now().UnixNano())),
		sleep: sleepContext}, nil
}

// Delay 返回第 retry 次重试前的等待时间，retry 从 1 开始。
// 基础延迟按 2 的幂增长并封顶，再乘以 [1-Jitter,1] 的随机
// 因子；抖动不改变重试次数，只把同时失败的客户端错开。
func (r *Retrier) Delay(retry int) time.Duration {
	if retry < 1 {
		retry = 1
	}
	delay := r.cfg.BaseDelay
	for i := 1; i < retry; i++ {
		if delay >= r.cfg.MaxDelay/2 {
			delay = r.cfg.MaxDelay
			break
		}
		delay *= 2
	}
	if delay > r.cfg.MaxDelay {
		delay = r.cfg.MaxDelay
	}
	if r.cfg.Jitter == 0 {
		return delay
	}
	return time.Duration(float64(delay) * (1 - r.cfg.Jitter*r.rnd.Float64()))
}

// Do 执行 fn，直到成功、遇到不可重试错误或次数耗尽。
func (r *Retrier) Do(ctx context.Context, fn func(context.Context) error) error {
	var last error
	for attempt := 1; attempt <= r.cfg.MaxAttempts; attempt++ {
		if err := ctx.Err(); err != nil {
			return err
		}
		last = fn(ctx)
		if last == nil {
			return nil
		}
		if !r.cfg.Retryable(last) {
			return last
		}
		if attempt == r.cfg.MaxAttempts {
			break
		}
		if err := r.sleep(ctx, r.Delay(attempt)); err != nil {
			return err
		}
	}
	return last
}
func sleepContext(ctx context.Context, d time.Duration) error {
	timer := time.NewTimer(d)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}

cd .local/checks/gs-practice-lab
gofmt -l ch14/breaker
go vet ./ch14/breaker/
go test ./ch14/breaker/ -count=1