《游戏服务端编程实践》5.2.3 动态转发与广播

// file: ch05/forward/broadcast.go
// tier: L1
// verify: go test ./ch05/forward/

package forward

import (
	"sync"
	"sync/atomic"
)

// Member 是一个在线玩家在 Hub 中的投递队列；队列满时丢弃并计数，绝不阻塞发送方。
type Member struct {
	PlayerID uint64
	RoomID   uint64
	queue    chan []byte
	dropped  atomic.Uint64
	closed   atomic.Bool
}

func (m *Member) Messages() <-chan []byte { return m.queue }
func (m *Member) Dropped() uint64         { return m.dropped.Load() }
func (m *Member) Closed() bool            { return m.closed.Load() }

func (m *Member) enqueue(msg []byte) bool {
	if m.closed.Load() {
		return false
	}
	select {
	case m.queue <- msg:
		return true
	default:
		m.dropped.Add(1)
		return false
	}
}

// Hub 维护 playerID 到 Member 的映射和房间成员表；广播先取快照，再在锁外逐条投递。
type Hub struct {
	mu      sync.RWMutex
	size    int
	members map[uint64]*Member
	rooms   map[uint64]map[uint64]struct{}
}

func NewHub(queueSize int) *Hub {
	if queueSize <= 0 {
		queueSize = 1
	}
	return &Hub{
		size:    queueSize,
		members: make(map[uint64]*Member),
		rooms:   make(map[uint64]map[uint64]struct{}),
	}
}

// Join 注册或顶替玩家连接；旧 Member 被标记关闭，写协程应尽快退出。
func (h *Hub) Join(playerID, roomID uint64) *Member {
	h.mu.Lock()
	defer h.mu.Unlock()
	if old, ok := h.members[playerID]; ok {
		old.closed.Store(true)
		h.removeFromRoomLocked(old)
	}
	m := &Member{PlayerID: playerID, RoomID: roomID, queue: make(chan []byte, h.size)}
	h.members[playerID] = m
	if roomID != 0 {
		if h.rooms[roomID] == nil {
			h.rooms[roomID] = make(map[uint64]struct{})
		}
		h.rooms[roomID][playerID] = struct{}{}
	}
	return m
}

func (h *Hub) Leave(playerID uint64) {
	h.mu.Lock()
	defer h.mu.Unlock()
	m, ok := h.members[playerID]
	if !ok {
		return
	}
	m.closed.Store(true)
	h.removeFromRoomLocked(m)
	delete(h.members, playerID)
}

func (h *Hub) removeFromRoomLocked(m *Member) {
	if m.RoomID == 0 {
		return
	}
	members := h.rooms[m.RoomID]
	delete(members, m.PlayerID)
	if len(members) == 0 {
		delete(h.rooms, m.RoomID)
	}
}

func (h *Hub) SendToPlayer(playerID uint64, msg []byte) bool {
	h.mu.RLock()
	m := h.members[playerID]
	h.mu.RUnlock()
	if m == nil {
		return false
	}
	return m.enqueue(msg)
}

func (h *Hub) BroadcastRoom(roomID uint64, msg []byte) int {
	h.mu.RLock()
	snapshot := make([]*Member, 0, len(h.rooms[roomID]))
	for playerID := range h.rooms[roomID] {
		if m := h.members[playerID]; m != nil {
			snapshot = append(snapshot, m)
		}
	}
	h.mu.RUnlock()
	sent := 0
	for _, m := range snapshot {
		if m.enqueue(msg) {
			sent++
		}
	}
	return sent
}

func (h *Hub) BroadcastAll(msg []byte) int {
	h.mu.RLock()
	snapshot := make([]*Member, 0, len(h.members))
	for _, m := range h.members {
		snapshot = append(snapshot, m)
	}
	h.mu.RUnlock()
	sent := 0
	for _, m := range snapshot {
		if m.enqueue(msg) {
			sent++
		}
	}
	return sent
}

// Dropped 返回在线玩家因队列满被丢弃的消息数；玩家已离线时第二个返回值为 false。
func (h *Hub) Dropped(playerID uint64) (uint64, bool) {
	h.mu.RLock()
	m := h.members[playerID]
	h.mu.RUnlock()
	if m == nil {
		return 0, false
	}
	return m.Dropped(), true
}

// file: ch05/forward/broadcast_test.go
// tier: L1
// verify: go test ./ch05/forward/

package forward

import "testing"

func TestSlowMemberDoesNotBlockOthers(t *testing.T) {
	hub := NewHub(1)
	slow := hub.Join(1, 10)
	fast := hub.Join(2, 10)
	if got := hub.BroadcastAll([]byte("m1")); got != 2 {
		t.Fatalf("first broadcast=%d, want 2", got)
	}
	if msg := <-fast.Messages(); string(msg) != "m1" {
		t.Fatalf("fast got %q, want m1", msg)
	}
	if got := hub.BroadcastAll([]byte("m2")); got != 1 {
		t.Fatalf("second broadcast=%d, want 1", got)
	}
	if slow.Dropped() != 1 || fast.Dropped() != 0 {
		t.Fatalf("dropped slow=%d fast=%d, want 1/0", slow.Dropped(), fast.Dropped())
	}
	if msg := <-slow.Messages(); string(msg) != "m1" {
		t.Fatalf("slow got %q, want queued m1", msg)
	}
	if msg := <-fast.Messages(); string(msg) != "m2" {
		t.Fatalf("fast got %q, want m2", msg)
	}
}

func TestQueueFullDropsAreCounted(t *testing.T) {
	hub := NewHub(1)
	m := hub.Join(1, 0)
	if !hub.SendToPlayer(1, []byte("a")) {
		t.Fatal("first send should be queued")
	}
	if hub.SendToPlayer(1, []byte("b")) {
		t.Fatal("second send should be dropped")
	}
	if hub.SendToPlayer(1, []byte("c")) {
		t.Fatal("third send should be dropped")
	}
	if n, ok := hub.Dropped(1); !ok || n != 2 {
		t.Fatalf("dropped=%d/%v, want 2/true", n, ok)
	}
	if got := len(m.Messages()); got != 1 {
		t.Fatalf("queued=%d, want 1", got)
	}
	if _, ok := hub.Dropped(99); ok {
		t.Fatal("offline player should not report drops")
	}
}

func TestBroadcastRoomIsBounded(t *testing.T) {
	hub := NewHub(4)
	p1 := hub.Join(1, 10)
	p2 := hub.Join(2, 10)
	p3 := hub.Join(3, 20)
	if got := hub.BroadcastRoom(10, []byte("room")); got != 2 {
		t.Fatalf("room broadcast=%d, want 2", got)
	}
	if got := len(p3.Messages()); got != 0 {
		t.Fatalf("outsider queued=%d, want 0", got)
	}
	if got := hub.BroadcastAll([]byte("all")); got != 3 {
		t.Fatalf("all broadcast=%d, want 3", got)
	}
	if got := len(p1.Messages()); got != 2 {
		t.Fatalf("p1 queued=%d, want room+all", got)
	}
	if msg := <-p1.Messages(); string(msg) != "room" {
		t.Fatalf("p1 first=%q, want room", msg)
	}
	if msg := <-p1.Messages(); string(msg) != "all" {
		t.Fatalf("p1 second=%q, want all", msg)
	}
	if got := hub.BroadcastRoom(99, []byte("x")); got != 0 {
		t.Fatalf("empty room broadcast=%d, want 0", got)
	}
	hub.Join(3, 10)
	if got := hub.BroadcastRoom(20, []byte("y")); got != 0 {
		t.Fatalf("old room broadcast=%d, want 0 after move", got)
	}
	hub.Leave(2)
	if got := hub.BroadcastRoom(10, []byte("z")); got != 2 {
		t.Fatalf("room broadcast after leave=%d, want 2", got)
	}
	_ = p2
}

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

ok  	gspractice/ch05/forward	0.565s