《游戏服务端编程实践》11.5 实战：服务间事件调度系统

// file: ch11/dispatcher/dispatcher.go
// tier: L1
// verify: go test ./ch11/dispatcher/

// Package dispatcher 提供按聚合 key 串行、带重试与死信的事件调度器。
package dispatcher

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

var (
	ErrEmptyKey   = errors.New("dispatcher: empty aggregate key")
	ErrNoHandler  = errors.New("dispatcher: no handler for topic")
	ErrBadHandler = errors.New("dispatcher: nil handler")
)

// Event 是调度器的最小输入。Key 是聚合键，例如 player:<id> 或 room:<id>。
type Event struct {
	ID      string
	Topic   string
	Key     string
	Payload []byte
}

// Handler 返回错误表示本次投递失败，由调度器决定重试或进死信。
type Handler func(ctx context.Context, ev Event) error

// Config 控制并发上限、重试上限和退避区间。
type Config struct {
	Workers       int
	MaxAttempts   int
	BaseBackoff   time.Duration
	MaxBackoff    time.Duration
	DeadLetterCap int
}

func (c Config) normalized() Config {
	if c.Workers <= 0 {
		c.Workers = 1
	}
	if c.MaxAttempts <= 0 {
		c.MaxAttempts = 1
	}
	if c.BaseBackoff < 0 {
		c.BaseBackoff = 0
	}
	if c.MaxBackoff < c.BaseBackoff {
		c.MaxBackoff = c.BaseBackoff
	}
	if c.DeadLetterCap <= 0 {
		c.DeadLetterCap = 1000
	}
	return c
}

// DeadLetter 保留最终失败的事件、尝试次数和最后一次错误。
type DeadLetter struct {
	Event    Event
	Attempts int
	Err      string
	At       time.Time
}

// Stats 是累计计数器，便于接指标和告警。
type Stats struct {
	Received     uint64
	Delivered    uint64
	Attempts     uint64
	Retried      uint64
	DeadLettered uint64
}

// lane 是一个聚合键的队列。running 保证同一时刻只有一个 worker 处理该 key。
type lane struct {
	key     string
	queue   []Event
	running bool
}

// Dispatcher 把事件路由到 topic handler，并按 key 串行执行。
type Dispatcher struct {
	cfg Config

	mu       sync.Mutex
	handlers map[string]Handler
	lanes    map[string]*lane
	active   int
	dead     []DeadLetter
	stats    Stats
	sleep    func(time.Duration)
	wg       sync.WaitGroup
}

// New 创建调度器。Workers 是同时执行的聚合键数量上限，不是事件吞吐上限。
func New(cfg Config) *Dispatcher {
	cfg = cfg.normalized()
	return &Dispatcher{
		cfg:      cfg,
		handlers: make(map[string]Handler),
		lanes:    make(map[string]*lane),
		sleep:    time.Sleep,
	}
}

// Register 注册 topic handler。重复注册同一 topic 会被拒绝，避免路由歧义。
func (d *Dispatcher) Register(topic string, h Handler) error {
	if topic == "" {
		return errors.New("dispatcher: empty topic")
	}
	if h == nil {
		return ErrBadHandler
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if _, ok := d.handlers[topic]; ok {
		return fmt.Errorf("dispatcher: topic %q already registered", topic)
	}
	d.handlers[topic] = h
	return nil
}

// Submit 把事件放入对应 key 的队列，并在有空闲 worker 时立即调度。
func (d *Dispatcher) Submit(ev Event) error {
	if ev.Key == "" {
		return ErrEmptyKey
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if _, ok := d.handlers[ev.Topic]; !ok {
		return ErrNoHandler
	}
	l := d.lanes[ev.Key]
	if l == nil {
		l = &lane{key: ev.Key}
		d.lanes[ev.Key] = l
	}
	l.queue = append(l.queue, ev)
	d.stats.Received++
	d.scheduleLocked()
	return nil
}

// scheduleLocked 在持锁状态下挑选未运行的 key 启动 worker。
func (d *Dispatcher) scheduleLocked() {
	for d.active < d.cfg.Workers {
		var next *lane
		for _, l := range d.lanes {
			if !l.running && len(l.queue) > 0 {
				next = l
				break
			}
		}
		if next == nil {
			return
		}
		next.running = true
		d.active++
		d.wg.Add(1)
		go d.runLane(next)
	}
}

// runLane 串行消费一个 key 的队列，消费完后释放 worker 给其他 key。
func (d *Dispatcher) runLane(l *lane) {
	defer d.wg.Done()
	for {
		d.mu.Lock()
		if len(l.queue) == 0 {
			l.running = false
			d.active--
			d.scheduleLocked()
			d.mu.Unlock()
			return
		}
		ev := l.queue[0]
		l.queue = l.queue[1:]
		d.mu.Unlock()
		d.deliver(ev)
	}
}

// deliver 在单个 key 内同步重试，失败到上限后写入死信。
func (d *Dispatcher) deliver(ev Event) {
	attempts := 0
	for {
		attempts++
		d.mu.Lock()
		h := d.handlers[ev.Topic]
		d.stats.Attempts++
		d.mu.Unlock()
		err := h(context.Background(), ev)
		if err == nil {
			d.mu.Lock()
			d.stats.Delivered++
			d.mu.Unlock()
			return
		}
		if attempts >= d.cfg.MaxAttempts {
			d.mu.Lock()
			d.stats.DeadLettered++
			d.dead = append(d.dead, DeadLetter{Event: ev, Attempts: attempts, Err: err.Error(), At: time.Now()})
			if len(d.dead) > d.cfg.DeadLetterCap {
				d.dead = append([]DeadLetter(nil), d.dead[len(d.dead)-d.cfg.DeadLetterCap:]...)
			}
			d.mu.Unlock()
			return
		}
		d.mu.Lock()
		d.stats.Retried++
		d.mu.Unlock()
		d.sleep(backoff(d.cfg, attempts))
	}
}

// backoff 返回第 attempt 次失败后的等待时间：base、2*base、4*base，直到上限。
func backoff(cfg Config, attempt int) time.Duration {
	delay := cfg.BaseBackoff
	for i := 1; i < attempt; i++ {
		if delay >= cfg.MaxBackoff/2 {
			return cfg.MaxBackoff
		}
		delay *= 2
	}
	if delay > cfg.MaxBackoff {
		return cfg.MaxBackoff
	}
	return delay
}

// Wait 等待当前已入队的事件全部处理完。它不停止调度器。
func (d *Dispatcher) Wait() {
	d.wg.Wait()
}

// Stats 返回计数器快照。
func (d *Dispatcher) Stats() Stats {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.stats
}

// DeadLetters 返回死信副本，调用方修改切片不会影响调度器。
func (d *Dispatcher) DeadLetters() []DeadLetter {
	d.mu.Lock()
	defer d.mu.Unlock()
	return append([]DeadLetter(nil), d.dead...)
}

// ReplayDeadLetters 把当前死信重新提交，返回成功入队的数量。
func (d *Dispatcher) ReplayDeadLetters() int {
	d.mu.Lock()
	batch := d.dead
	d.dead = nil
	d.mu.Unlock()
	n := 0
	for _, dl := range batch {
		if err := d.Submit(dl.Event); err == nil {
			n++
		}
	}
	return n
}

cd .local/checks/gs-practice-lab && gofmt -l ch11/dispatcher && go vet ./ch11/dispatcher/ && go test ./ch11/dispatcher/ -count=1

ok  	gspractice/ch11/dispatcher	0.331s