《MMO 服务器开发》8.3 广播与扇出

for conn in conns:
    buf = EncodeFrame(cmd, seq, flags, payload)   # 每个接收者都编码一次
    conn.WriteFrame(buf)                           # 每个接收者都写一次

// file: internal/router/fanout.go
// tier: L1
// verify: go test ./internal/router/

package router

import "sync"

// WireConn accepts an already-encoded frame. A real session.Session from
// chapter 7 satisfies it; tests and benchmarks use a counting fake. Working at
// the byte level here is deliberate: a fanout path must be able to encode once
// and reuse the same buffer for every recipient.
type WireConn interface {
	ID() uint64
	WriteFrame(buf []byte) error
}

// Broadcaster fans one logical message out to a set of connections. The mutex
// guards only the membership map, never the writes: a broadcast takes a snapshot
// under the read lock and then writes outside it, so one stuck socket cannot
// stall Add/Remove or the next broadcast.
type Broadcaster struct {
	mu    sync.RWMutex
	conns map[uint64]WireConn
}

// NewBroadcaster returns an empty broadcaster.
func NewBroadcaster() *Broadcaster {
	return &Broadcaster{conns: make(map[uint64]WireConn)}
}

// Add registers a connection.
func (b *Broadcaster) Add(c WireConn) {
	b.mu.Lock()
	b.conns[c.ID()] = c
	b.mu.Unlock()
}

// Remove drops a connection by id. Removing an absent id is a no-op.
func (b *Broadcaster) Remove(id uint64) {
	b.mu.Lock()
	delete(b.conns, id)
	b.mu.Unlock()
}

// Count reports how many connections are registered.
func (b *Broadcaster) Count() int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	return len(b.conns)
}

// snapshot copies the recipient slice under the read lock. The writes that
// follow hold no lock at all.
func (b *Broadcaster) snapshot() []WireConn {
	b.mu.RLock()
	defer b.mu.RUnlock()
	out := make([]WireConn, 0, len(b.conns))
	for _, c := range b.conns {
		out = append(out, c)
	}
	return out
}

// BroadcastEncoded encodes the frame exactly once and writes the same bytes to
// every recipient. Encoding cost is O(1) in the number of recipients; only the
// socket writes scale with N. This is the shape every hot fanout path should
// have.
func (b *Broadcaster) BroadcastEncoded(cmd Cmd, seq uint32, flags uint8, payload []byte) (sent int) {
	buf := EncodeFrame(nil, cmd, seq, flags, payload)
	for _, c := range b.snapshot() {
		if err := c.WriteFrame(buf); err == nil {
			sent++
		}
	}
	return sent
}

// BroadcastNaive re-encodes the frame once per recipient. It exists only so the
// benchmark can quantify what BroadcastEncoded saves; production code should
// never call it.
func (b *Broadcaster) BroadcastNaive(cmd Cmd, seq uint32, flags uint8, payload []byte) (sent int) {
	for _, c := range b.snapshot() {
		buf := EncodeFrame(nil, cmd, seq, flags, payload)
		if err := c.WriteFrame(buf); err == nil {
			sent++
		}
	}
	return sent
}

// BatchBuffer concatenates several encoded frames into one buffer so a caller
// can hand the socket a single write (a gather-write stand-in). Batching turns
// K syscalls into one and is the second lever, after encode-once, for a hot
// push path.
type BatchBuffer struct {
	buf []byte
	n   int
}

// NewBatchBuffer returns a buffer with room for capBytes of encoded frames.
func NewBatchBuffer(capBytes int) *BatchBuffer {
	if capBytes < 0 {
		capBytes = 0
	}
	return &BatchBuffer{buf: make([]byte, 0, capBytes)}
}

// Add appends one encoded frame.
func (bb *BatchBuffer) Add(cmd Cmd, seq uint32, flags uint8, payload []byte) {
	bb.buf = EncodeFrame(bb.buf, cmd, seq, flags, payload)
	bb.n++
}

// Bytes returns the concatenated frames.
func (bb *BatchBuffer) Bytes() []byte { return bb.buf }

// Len reports how many frames are buffered.
func (bb *BatchBuffer) Len() int { return bb.n }

// Reset clears the buffer but keeps its backing array, so the next batch reuses
// the same allocation.
func (bb *BatchBuffer) Reset() {
	bb.buf = bb.buf[:0]
	bb.n = 0
}

b.mu.Lock()
for _, c := range b.conns { c.WriteFrame(buf) }   // 写发生在锁内
b.mu.Unlock()

持锁写（慢客户端拖死全体）        快照 + 锁外写（本节做法）
───────────────────────────       ─────────────────────────────
Lock()                            RLock() → 拷贝接收者切片 → RUnlock()
  for c { c.Write(buf) }  ←阻塞    for c { c.Write(buf) }   ← 锁外
Unlock()                          慢的只慢自己，其余广播与上下线不受影响

// snippet: internal/router/fanout_test.go（节选）

// 1000 个连接，128 字节 payload，两个实现跑同一个基准。
func BenchmarkBroadcastEncoded(b *testing.B) {
	bd := NewBroadcaster()
	for i := 0; i < 1000; i++ {
		bd.Add(&countingConn{id: uint64(i)})
	}
	payload := make([]byte, 128)
	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		bd.BroadcastEncoded(CmdAttrSync, uint32(i), 0, payload)
	}
}

$ go test ./internal/router/ -run '^$' -bench 'BenchmarkBroadcast' -benchmem -count=3
BenchmarkBroadcastEncoded-10   64026   16480 ns/op   16544 B/op   3 allocs/op
BenchmarkBroadcastNaive-10     20070   67489 ns/op  176384 B/op   2001 allocs/op