《游戏服务端架构·规模化、全球化与成本》6.2 积压、顺序与死信

pending -> handling -> acked
             |
             | error, attempts < max
             v
          retry(head)
             |
             | attempts = max
             v
        dead-letter -> 人工重放（必须幂等） -> acked

// file: internal/backlogorder/backlogorder.go
// tier: L1
// verify: go test ./internal/backlogorder/
package backlogorder

import (
	"errors"
	"hash/fnv"
	"sync"
	"time"
)

// Record 是分区队列中的一条待处理记录。
type Record struct {
	Offset     int64
	Key        string
	Payload    []byte
	EnqueuedAt time.Time
	Attempts   int
}

// PartitionOf 用 FNV-1a 把业务键映射到固定分区，同一键永远落在同一分区。
func PartitionOf(key string, partitions int) int {
	if partitions < 1 {
		partitions = 1
	}
	h := fnv.New32a()
	_, _ = h.Write([]byte(key))
	return int(h.Sum32() % uint32(partitions))
}

// DeadLetter 保存被隔离的毒消息及其失败原因。
type DeadLetter struct {
	Record   Record
	Reason   string
	FailedAt time.Time
}

// Queue 是内存分区队列，用锁模拟每个分区只有一个消费者。
type Queue struct {
	mu         sync.Mutex
	partitions [][]Record
	dead       []DeadLetter
}

// New 创建固定数量的分区；分区数一旦确定不可在线增减，否则键的归属会漂移。
func New(partitions int) *Queue {
	if partitions < 1 {
		partitions = 1
	}
	return &Queue{partitions: make([][]Record, partitions)}
}

// Publish 返回记录落入的分区号，便于测试与埋点。
func (q *Queue) Publish(r Record) int {
	p := PartitionOf(r.Key, len(q.partitions))
	q.mu.Lock()
	defer q.mu.Unlock()
	q.partitions[p] = append(q.partitions[p], r)
	return p
}

// Drain 按顺序消费单个分区。失败记录在队首重试；达到上限后进入死信，
// 分区继续前进，毒消息不会永久阻塞后面的记录。
func (q *Queue) Drain(partition, maxAttempts int, handle func(Record) error) (delivered, dead int) {
	if maxAttempts < 1 {
		maxAttempts = 1
	}
	q.mu.Lock()
	defer q.mu.Unlock()
	if partition < 0 || partition >= len(q.partitions) {
		return 0, 0
	}
	pending := q.partitions[partition]
	q.partitions[partition] = nil
	for len(pending) > 0 {
		rec := pending[0]
		if err := handle(rec); err != nil {
			rec.Attempts++
			if rec.Attempts >= maxAttempts {
				q.dead = append(q.dead, DeadLetter{
					Record:   rec,
					Reason:   err.Error(),
					FailedAt: time.Now(),
				})
				pending = pending[1:]
				dead++
				continue
			}
			pending[0] = rec
			// 本轮到此为止：保持顺序，等调用方修复原因后再次 Drain。
			break
		}
		pending = pending[1:]
		delivered++
	}
	q.partitions[partition] = pending
	return delivered, dead
}

// Backlog 是一次积压快照。条数、字节数、最老消息年龄回答三个不同问题。
type Backlog struct {
	Count     int
	Bytes     int
	OldestAge time.Duration
}

// Backlog 汇总全部分区的积压。
func (q *Queue) Backlog(now time.Time) Backlog {
	q.mu.Lock()
	defer q.mu.Unlock()
	var b Backlog
	for _, pending := range q.partitions {
		for _, rec := range pending {
			b.Count++
			b.Bytes += len(rec.Payload)
			age := now.Sub(rec.EnqueuedAt)
			if age > b.OldestAge {
				b.OldestAge = age
			}
		}
	}
	return b
}

// AlertPolicy 以最老消息年龄为主判据，条数只做辅助告警。
type AlertPolicy struct {
	WarnAge  time.Duration
	PageAge  time.Duration
	MaxCount int
}

// Evaluate 返回 ok、warn 或 page。
func (p AlertPolicy) Evaluate(b Backlog) string {
	switch {
	case p.PageAge > 0 && b.OldestAge >= p.PageAge:
		return "page"
	case p.WarnAge > 0 && b.OldestAge >= p.WarnAge:
		return "warn"
	case p.MaxCount > 0 && b.Count >= p.MaxCount:
		return "warn"
	default:
		return "ok"
	}
}

// DeadLetters 返回死信副本。
func (q *Queue) DeadLetters() []DeadLetter {
	q.mu.Lock()
	defer q.mu.Unlock()
	out := make([]DeadLetter, len(q.dead))
	copy(out, q.dead)
	return out
}

var errEmptyDeadLetter = errors.New("backlogorder: empty dead-letter id")

// Replay 重放死信。applied 返回 true 表示业务侧已处理过，必须跳过；
// 成功重放的记录从死信列表移除，失败的保留供人工介入。
func (q *Queue) Replay(applied func(Record) bool, handle func(Record) error) (replayed, skipped int, err error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	var remain []DeadLetter
	for _, dl := range q.dead {
		if dl.Record.Key == "" {
			err = errEmptyDeadLetter
			remain = append(remain, dl)
			continue
		}
		if applied != nil && applied(dl.Record) {
			skipped++
			continue
		}
		if handle != nil {
			if herr := handle(dl.Record); herr != nil {
				dl.Reason = herr.Error()
				remain = append(remain, dl)
				continue
			}
		}
		replayed++
	}
	q.dead = remain
	return replayed, skipped, err
}

cd .local/checks/gsarch-lab
go test -count=1 ./internal/backlogorder/
ok  gsarchlab/internal/backlogorder