《游戏服务端架构·核心系统与契约》2.3 登录排队、限流与开服洪峰

// file: internal/loginqueue/pump.go
// tier: L1
// verify: go test ./internal/loginqueue/

// Package loginqueue combines a token bucket with a fair, tenant-aware waiting
// queue. It separates "may I enter now" from "where am I in line".
package loginqueue

import (
	"math"
	"time"
)

// Pump admits at most max waiting tickets, round-robin across tenants.
func (q *Queue) Pump(max int) []Ticket {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.expireLocked(q.now())
	want := len(q.order)
	if max > 0 && max < want {
		want = max
	}
	if want == 0 {
		return nil
	}
	granted := q.limiter.Grant(want)
	out := make([]Ticket, 0, granted)
	for len(out) < granted {
		id := q.nextLocked()
		if id == "" {
			break
		}
		t := q.byID[id]
		t.State = Admitted
		q.removeLocked(id)
		out = append(out, *t)
	}
	return out
}

// nextLocked pops the front tenant, takes one ticket, then rotates that
// tenant to the back if it still has waiters. Rotating explicitly (rather
// than indexing a sorted list) keeps fairness stable as tenants come and go.
func (q *Queue) nextLocked() string {
	for len(q.ring) > 0 {
		name := q.ring[0]
		q.ring = q.ring[1:]
		ids := q.tenants[name]
		if len(ids) == 0 {
			delete(q.tenants, name)
			continue
		}
		id := ids[0]
		if len(ids) == 1 {
			delete(q.tenants, name)
		} else {
			q.tenants[name] = ids[1:]
			q.ring = append(q.ring, name)
		}
		return id
	}
	return ""
}

func (q *Queue) dropRingLocked(name string) {
	for idx, v := range q.ring {
		if v == name {
			q.ring = append(q.ring[:idx], q.ring[idx+1:]...)
			return
		}
	}
}

func (q *Queue) removeLocked(id string) {
	for idx, v := range q.order {
		if v == id {
			q.order = append(q.order[:idx], q.order[idx+1:]...)
			break
		}
	}
	for name, ids := range q.tenants {
		for idx, v := range ids {
			if v != id {
				continue
			}
			ids = append(ids[:idx], ids[idx+1:]...)
			if len(ids) == 0 {
				delete(q.tenants, name)
				q.dropRingLocked(name)
			} else {
				q.tenants[name] = ids
			}
			break
		}
	}
}

func (q *Queue) expireLocked(nowT time.Time) {
	for _, id := range append([]string(nil), q.order...) {
		t := q.byID[id]
		if t == nil || nowT.Before(t.Deadline) {
			continue
		}
		t.State = Expired
		q.removeLocked(id)
	}
}

func (q *Queue) snapshotLocked(t *Ticket) Ticket {
	cp := *t
	if cp.State != Waiting {
		return cp
	}
	cp.Position = 0
	for idx, id := range q.order {
		if id == t.ID {
			cp.Position = idx + 1
			break
		}
	}
	cp.Wait = q.estimateLocked(cp.Position)
	return cp
}

func (q *Queue) estimateLocked(position int) time.Duration {
	if position <= 0 {
		return 0
	}
	rate := q.limiter.Rate()
	if rate <= 0 {
		return time.Duration(math.MaxInt64)
	}
	need := float64(position) - q.limiter.Available()
	if need <= 0 {
		return 0
	}
	return time.Duration(need / rate * float64(time.Second))
}

// file: internal/loginqueue/expire_test.go
// tier: L1
package loginqueue

import (
	"testing"
	"time"
)

func TestExpireAndSweep(t *testing.T) {
	clk := newClock()
	q, err := NewQueue(NewLimiter(1, 1, clk.Now), 5*time.Second, clk.Now)
	if err != nil {
		t.Fatalf("NewQueue: %v", err)
	}
	expired := q.Enqueue("A", "r1")
	clk.Advance(6 * time.Second)
	if poll := q.Poll(expired.ID); poll.State != Expired {
		t.Fatalf("state = %v, want Expired", poll.State)
	}
	if q.Len() != 0 {
		t.Fatalf("expired ticket still occupies the queue: %d", q.Len())
	}
	fresh := q.Enqueue("A", "r2")
	admitted := q.Pump(1)
	if len(admitted) != 1 || admitted[0].ID != fresh.ID {
		t.Fatalf("expiry did not free the grant: %+v", admitted)
	}
	clk.Advance(10 * time.Second)
	if n := q.Sweep(time.Second); n < 1 {
		t.Fatalf("swept %d tickets, want at least 1", n)
	}
	if poll := q.Poll(fresh.ID); poll.State != Unknown {
		t.Fatalf("swept ticket state = %v, want Unknown", poll.State)
	}
}

$ cd .local/checks/gsarch-lab
$ gofmt -l internal/loginqueue
$ go vet ./internal/loginqueue/...
$ go test -count=1 ./internal/loginqueue/...
ok  	gsarchlab/internal/loginqueue