《游戏服务端编程实践》10.1.2 消息队列与转发

一、一条聊天消息的完整路径

聊天不是“收到就广播”。从网关收到字节到目标玩家看到文字，中间至少有五段职责，每段失败时应该返回不同错误，而不是统一回一句“发送失败”。

阶段拆开的价值在于可观测性。限频拒绝应该记在网关指标上，敏感词命中应该进审核队列，投递失败应该进重试或离线补发。全部混在一个函数里，线上只能看到“聊天失败总数”，无法定位是刷屏、违规还是网络问题。

另一个收益是重试边界清晰。校验和过滤是纯计算，失败直接返回错误即可；路由失败可能只是成员列表短暂为空，值得重试；投递失败则要区分连接断开和队列满，前者转离线，后者按策略丢弃。把可重试和不可重试的阶段分开，重试逻辑才不会把一条违规消息反复送进过滤器。

二、同步转发还是队列异步

最直接的实现是收到消息后同步遍历成员并写连接。它在小规模下工作良好，但把网络写入、数据库写入和业务校验绑在同一个调用栈里，任何一个慢操作都会放大成接入层超时。

异步化不是免费的。消息进入队列后，发送者拿到的“成功”只代表入队成功，不代表已投递。产品上的“发送成功”提示要与这个语义对齐，否则玩家会以为对方一定收到了。对系统公告这类必须送达的消息，入队前先落库，再由消费者重试，才能做到重启不丢。

还有一类消息不适合异步：玩家在副本里发出的战术指令。它需要和移动、技能指令共享顺序，走聊天队列反而可能晚于战斗帧到达。这类内容应该并入战斗消息通道，而不是复用世界聊天的基础设施。

三、消息 ID、顺序与去重

消息 ID 至少承担三个职责：客户端去重、服务端排序、问题追踪。ID 可以由接入层生成全局递增号，也可以由聊天服务按频道生成序号。两者不要混用：全局 ID 适合日志检索，频道序号适合判断消息之间有没有缺口。

频道内顺序应该由服务端决定。客户端并发发送两条消息，到达顺序本来就不确定；如果服务端再为每条消息启动独立协程，最终投递顺序可能与接收顺序完全相反。正确做法是同一频道的消息进入同一队列，由单个消费者串行处理。跨频道不需要顺序，可以并行。

去重通常分两层：客户端重发时带同一个客户端消息 ID，服务端在短窗口内拒绝重复；服务端投递时带服务端消息 ID，客户端按 ID 去重。只做其中一层都不够，尤其是弱网重连后客户端重发非常常见。

去重窗口不能无限增长。通常按“玩家 ID + 客户端消息 ID”保留最近几分钟，或者每个玩家保留固定数量的最近 ID；超出窗口后允许重复，由客户端的服务端消息 ID 去重兜底。窗口大小要和客户端重试时长匹配，客户端重试 30 秒，服务端只保留 5 秒，重复消息就会重新进入频道。

四、跨网关投递

玩家分散在多个网关进程上，聊天服务需要知道“某个玩家当前挂在哪条连接上”。这张在线路由表通常由网关在连接建立和断开时上报，聊天服务只读。投递时按目标网关分组，一次 RPC 发送一批消息，而不是每个玩家一次调用。

跨网关投递有三种常见失败：目标网关刚刚断开、RPC 超时但对方其实已处理、路由表短暂过期。第一种直接跳过并转离线；第二种要求投递接口幂等，接收端按消息 ID 去重；第三种会在重连后被修正，但期间的消息可能投错连接，所以连接会话必须有唯一代号，不能只用玩家 ID。

公会和世界频道还要处理“成员很多、在线很少”的情况。路由层应该先取在线集合，再按网关分组，避免为每个离线成员构造一次无意义调用。离线成员是否补发，取决于频道类型：公会可以补，世界频道一般不补。

跨网关批量投递还要注意“部分成功”。一批 100 条消息发给某个网关，返回结果可能是 60 条成功、40 条连接已断。调用方不能把整批标记为失败后全部重发，否则成功的 60 条会重复；也不能只标记批次成功，否则断开的 40 条永远丢失。正确做法是让投递结果按消息 ID 返回，成功的不再重试，失败的转入离线或死信。

五、背压与丢弃策略

队列必须有界。无界队列在流量突增时不会立刻报错，而是把内存吃满，最后整个进程被系统杀掉，连战斗和登录一起受影响。有界队列把问题提前暴露成 ErrQueueFull，调用方可以选择拒绝、降级或丢弃旧消息。

“聊天可丢、公告不可丢”要落实到代码路径，而不是口头约定。最简单的做法是两类消息使用不同队列和不同容量，公告队列满时直接告警并停止接收新公告，避免静默丢失。

背压还要落到发送者维度。单个玩家每秒发 50 条世界消息，即使队列有界，也会迅速把容量占满，让其他玩家的正常消息被拒绝。应该先在接入层做按玩家的令牌桶，再进入频道队列；令牌桶拒绝的消息不占队列容量，也不会影响其他玩家。频道级限流则用于防止大量玩家同时刷屏，两层限流解决的是不同问题。

六、优先级与队列隔离

只有一个全局队列时，世界频道的闲聊会和系统公告争抢同一个缓冲区和同一批消费者。活动期间闲聊流量可以把公告挤到队尾，等公告被处理时，活动可能已经结束。解决办法不是给队列排序，而是按消息等级拆成独立队列，各自有容量、消费者和丢弃策略。

队列隔离后，世界频道再怎么刷屏，也只能打满自己的队列，不会影响私聊和公告。代价是资源需要分开核算：尽力队列可以多开消费者，可靠队列要限制并发以保护存储，公告队列则宁可慢也不能丢。这些参数应该来自容量评估，而不是所有队列复制同一个默认值。

隔离也会带来新的运维问题：队列太多，消费者和协程数量随之上涨；队列太少，又回到互相挤占。经验上按“丢弃策略”分组即可，同一组内再按频道分片。公告、可靠、尽力三类已经能覆盖大部分游戏聊天场景，不需要为每个业务活动单独建队列。

还要给每个队列记录三个指标：入队速率、出队速率、拒绝次数。只看 CPU 无法判断是消费者慢还是流量真的涨了；拒绝次数持续上升，才说明容量或丢弃策略需要调整。

七、失败场景：每条消息一个 goroutine

某次更新后，聊天服务为每条消息执行 go deliver(msg)。平时看不出问题，活动期间世界频道每秒几千条消息，goroutine 数量瞬间上涨，调度开销超过实际工作。更严重的是顺序：同一个玩家先收到“开团了”，再收到“团已满”，界面状态直接错乱。

这个问题有三个根因：没有队列、没有顺序约束、没有背压。修法不是给 goroutine 加锁，而是回到“频道内串行、频道间并行”的模型。每个频道一个消费者，消费者按 FIFO 取消息；频道数量再多，活跃消费者也只与有消息的频道相关。

还要注意修复顺序。先加有界队列，能立刻避免 goroutine 爆炸；再把同频道消息路由到同一个消费者，恢复顺序；最后补拒绝指标和按玩家限流。三步反过来做，中间态可能既没有顺序保证，也没有可观测数据，很难判断是否真的变好。

八、Go 实现：按频道分片串行，跨频道并行

ChatQueue 为每个频道惰性创建一个带缓冲的 channel 和一个消费协程。Enqueue 在锁内检查关闭状态并尝试非阻塞写入，队列满立即返回 ErrQueueFull；Close 关闭所有 done 信号并等待消费协程退出。

// file: ch10/chat/queue.go
// tier: L1
// verify: go test ./ch10/chat/

package chat

import (
	"errors"
	"sync"
)

var (
	ErrQueueFull   = errors.New("chat: queue full")
	ErrQueueClosed = errors.New("chat: queue closed")
)

// Envelope 是进入队列的聊天消息，ID 由接入层分配，频道内必须保持递增。
type Envelope struct {
	ID        uint64
	ChannelID string
	From      uint64
	Text      string
}

type DeliveryFunc func(Envelope)

type shard struct {
	queue chan Envelope
	done  chan struct{}
	once  sync.Once
}

// ChatQueue 为每个频道维护一个独立队列和一个消费协程：
// 频道内串行保证顺序，频道之间并行避免互相阻塞。
type ChatQueue struct {
	mu       sync.RWMutex
	capacity int
	deliver  DeliveryFunc
	shards   map[string]*shard
	closed   bool
	wg       sync.WaitGroup
}

func NewChatQueue(capacity int, deliver DeliveryFunc) *ChatQueue {
	if capacity < 0 {
		capacity = 0
	}
	return &ChatQueue{
		capacity: capacity,
		deliver:  deliver,
		shards:   make(map[string]*shard),
	}
}

// getShard 必须在持有 q.mu 时调用，首次见到频道才创建队列和协程。
func (q *ChatQueue) getShard(channelID string) *shard {
	if s, ok := q.shards[channelID]; ok {
		return s
	}
	s := &shard{queue: make(chan Envelope, q.capacity), done: make(chan struct{})}
	q.shards[channelID] = s
	q.wg.Add(1)
	go q.run(s)
	return s
}

func (q *ChatQueue) run(s *shard) {
	defer q.wg.Done()
	for {
		select {
		case env := <-s.queue:
			q.deliver(env)
		case <-s.done:
			return
		}
	}
}

// Enqueue 不阻塞：队列满时立即返回 ErrQueueFull，由调用方决定丢弃旧消息还是拒绝新消息。
func (q *ChatQueue) Enqueue(env Envelope) error {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.closed {
		return ErrQueueClosed
	}
	s := q.getShard(env.ChannelID)
	select {
	case s.queue <- env:
		return nil
	default:
		return ErrQueueFull
	}
}

// Close 停止所有消费协程并等待它们退出；正在执行的 deliver 不会被中断。
func (q *ChatQueue) Close() {
	q.mu.Lock()
	if q.closed {
		q.mu.Unlock()
		return
	}
	q.closed = true
	for _, s := range q.shards {
		s.once.Do(func() { close(s.done) })
	}
	q.mu.Unlock()
	q.wg.Wait()
}

func (q *ChatQueue) Pending(channelID string) int {
	q.mu.RLock()
	defer q.mu.RUnlock()
	s, ok := q.shards[channelID]
	if !ok {
		return 0
	}
	return len(s.queue)
}

func (q *ChatQueue) ChannelCount() int {
	q.mu.RLock()
	defer q.mu.RUnlock()
	return len(q.shards)
}

getShard 必须在持锁时调用，否则两个协程可能同时为同一个频道创建消费者，顺序保证立刻失效。Enqueue 使用 select 加 default 实现非阻塞写入，慢消费者不会把接入协程拖住。

Close 只停止消费者，不承诺把缓冲区里的消息全部处理完。因此可靠消息应当在入队前落库，或在关闭时先停止接收、再排空队列。把“关闭即丢”写在接口注释里，比事后猜行为更安全。

分片创建也是惰性的：没有消息的频道不会占用协程。代价是频道第一次收到消息时要承担创建开销，所以不要在投递热路径之外反复创建同名频道，否则 map 会不断膨胀。频道生命周期应由上层关闭逻辑管理，队列只负责消息本身。

九、测试覆盖哪些边界

queue_test.go 用四个测试固定队列契约：

• 同一频道连续入队 200 条，消费顺序必须与入队 ID 完全一致；两个频道交错入队也不互相影响。
• 容量为 1 且消费者被阻塞时，继续入队最终必须返回 ErrQueueFull，不能无限缓冲。
• 两个频道的消费者可以同时开始执行；如果实现退化成单消费者，测试会在超时后失败。
• Close 之后再次 Enqueue 返回 ErrQueueClosed，重复调用 Close 不会 panic。

这些测试没有覆盖真实网络写入，deliver 在测试里只是一个内存回调。接入线上时建议再加两类验证：一是用 -race 跑并发入队和关闭，确认没有数据竞争；二是注入一个随机阻塞的 deliver，观察队列满时调用方是否按预期执行丢弃或拒绝。后者比单纯的单元测试更接近活动期间的慢连接场景。

十、运行测试

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

ok  	gspractice/ch10/chat	0.233s

十一、接到线上前还要补什么

这段队列代码只解决“频道内保序、频道间并行、有界背压”三个问题，直接上线还缺几块。

第一是消息落库。可靠队列的消费者应该先写存储再投递，或者在入队前同步落库；当前 deliver 只是一个回调，存储失败时无法回滚入队动作，需要调用方保证幂等或补一个重试表。第二是优雅关闭。Close 会立刻停止消费，缓冲区里的消息可能没被处理；发布时应该先停止接收新消息，等待队列排空，再调用 Close，并给等待设置超时。

第三是可观测性。至少要记录每个频道的队列长度、拒绝次数和消费延迟。队列长度持续上涨说明消费者跟不上；拒绝次数上涨说明容量不足或有人在刷屏；消费延迟上涨通常意味着下游投递变慢。把这三个指标和频道 ID 关联，才能快速判断是单个公会出现问题，还是全局流量异常。

第四是消费者隔离。当前每个频道一个协程，频道数量受在线公会影响，极端情况下协程数仍然不少。可以在频道空闲一段时间后回收分片，或者把冷频道合并到共享 worker，但要保证同一频道同一时刻只有一个消费者，否则顺序保证会被破坏。

小结

聊天消息路径要拆成接入、校验、过滤、路由和投递五段，每段有自己的错误和指标。同步转发适合小房间，规模化后应按频道分片：频道内串行保序，频道间并行提吞吐，有界队列提供背压。消息 ID 负责去重和追踪，频道序号负责判断缺口；跨网关投递要按网关批量发送，并保证接收端幂等。世界频道可以丢旧消息，系统公告必须落库重试，这条差异要写进代码而不是只写在文档里。

常见错误与边界

• 每条消息启动一个 goroutine，流量上涨后调度开销爆炸，频道内顺序也无法保证。
• 队列无界，突发流量先把内存吃满，最终整个进程被系统杀掉。
• 同一频道使用多个消费者并行处理，后发的消息先到，界面状态错乱。
• 把“入队成功”当成“投递成功”提示玩家，对方离线时消息其实已经丢了。
• 跨网关投递不幂等，RPC 超时重试导致接收端收到两条相同消息。
• 在线路由只按玩家 ID 不按连接代号，重连后被投到旧连接。
• 系统公告和世界闲聊共用队列，闲聊刷屏挤掉公告且没有告警。
• Close 后仍允许入队，或者向已关闭的 channel 发送导致 panic。
• 频道首次入队才创建分片，却从不回收，长期运行后 map 和协程持续增长。
• 只在聊天服务做去重，跨网关重试后同一条消息仍可能被投递两次。