3.7 房间广播、弱网与国内网络环境

游戏服务器中，房间广播是高频操作（MOBA每秒10次），弱网环境是手游的常态（地铁、电梯），国内网络环境复杂（跨地域、多运营商）。

房间广播优化

批量广播

// RoomBroadcaster 房间广播器
type RoomBroadcaster struct {
    // 房间玩家
    players     map[uint64]*Player

    // 广播队列（批量发送）
    broadcastQueue chan *BroadcastTask

    // 批量大小
    batchSize    int
}

type BroadcastTask struct {
    Message     *Message
    Exclude     map[uint64]bool  // 排除的玩家
}

// Broadcast 广播消息
func (rb *RoomBroadcaster) Broadcast(msg *Message, exclude ...uint64) {
    // 创建排除列表
    excludeMap := make(map[uint64]bool)
    for _, id := range exclude {
        excludeMap[id] = true
    }

    // 添加到广播队列
    task := &BroadcastTask{
        Message: msg,
        Exclude: excludeMap,
    }

    select {
    case rb.broadcastQueue <- task:
        // 成功加入队列
    default:
        // 队列满，丢弃或等待
        log.Printf("Broadcast queue full")
    }
}

// 批量广播协程
func (rb *RoomBroadcaster) broadcastWorker() {
    batch := make([]*BroadcastTask, 0, rb.batchSize)

    ticker := time.NewTicker(10 * time.Millisecond)  // 每10ms批量发送
    defer ticker.Stop()

    for {
        select {
        case task := <-rb.broadcastQueue:
            batch = append(batch, task)

            // 达到批量大小，立即发送
            if len(batch) >= rb.batchSize {
                rb.flushBatch(batch)
                batch = batch[:0]
            }

        case <-ticker.C:
            // 定时发送
            if len(batch) > 0 {
                rb.flushBatch(batch)
                batch = batch[:0]
            }
        }
    }
}

// flushBatch 批量发送
func (rb *RoomBroadcaster) flushBatch(batch []*BroadcastTask) {
    // 按玩家分组消息
    playerMessages := make(map[uint64][]*Message)

    for _, task := range batch {
        for playerID := range rb.players {
            // 检查是否排除
            if task.Exclude[playerID] {
                continue
            }

            // 添加到玩家消息列表
            playerMessages[playerID] = append(playerMessages[playerID], task.Message)
        }
    }

    // 批量发送给每个玩家
    for playerID, messages := range playerMessages {
        player := rb.players[playerID]

        // 合并消息
        merged := rb.mergeMessages(messages)

        // 发送
        player.Send(merged)
    }
}

// mergeMessages 合并多条消息
func (rb *RoomBroadcaster) mergeMessages(messages []*Message) *Message {
    // 简化版：只保留最后一条消息
    // 实际中可以根据消息类型智能合并
    if len(messages) == 0 {
        return nil
    }

    return messages[len(messages)-1]
}

增量更新

// DeltaUpdate 增量更新
type DeltaUpdate struct {
    // 完整状态
    FullState   *GameState

    // 上次状态
    LastState   *GameState

    // 变化部分
    Deltas      []Delta
}

type Delta struct {
    EntityID    uint64
    Property    string
    OldValue    interface{}
    NewValue    interface{}
}

// CalculateDeltas 计算增量
func (du *DeltaUpdate) CalculateDeltas() []Delta {
    deltas := make([]Delta, 0)

    // 遍历所有实体
    for entityID, entity := range du.FullState.Entities {
        lastEntity, ok := du.LastState.Entities[entityID]
        if !ok {
            // 新实体，发送完整状态
            deltas = append(deltas, Delta{
                EntityID: entityID,
                Property: "full",
                OldValue: nil,
                NewValue: entity,
            })
            continue
        }

        // 对比属性变化
        if entity.Position != lastEntity.Position {
            deltas = append(deltas, Delta{
                EntityID: entityID,
                Property: "position",
                OldValue: lastEntity.Position,
                NewValue: entity.Position,
            })
        }

        if entity.HP != lastEntity.HP {
            deltas = append(deltas, Delta{
                EntityID: entityID,
                Property: "hp",
                OldValue: lastEntity.HP,
                NewValue: entity.HP,
            })
        }
    }

    return deltas
}

// BroadcastDeltas 广播增量更新
func (rb *RoomBroadcaster) BroadcastDeltas(delta *DeltaUpdate) {
    // 1. 计算增量
    deltas := delta.CalculateDeltas()

    // 2. 如果变化太大，发送完整状态
    if len(deltas) > 100 {
        rb.Broadcast(&Message{
            Type: MsgType_FullState,
            Body: delta.FullState.Serialize(),
        })
        return
    }

    // 3. 发送增量更新
    rb.Broadcast(&Message{
        Type: MsgType_DeltaUpdate,
        Body: serializeDeltas(deltas),
    })
}

弱网优化

断线重连

// ReconnectManager 断线重连管理器
type ReconnectManager struct {
    // 连接状态
    connected   bool

    // 重连配置
    maxRetries  int
    retryDelay  time.Duration

    // 状态缓存
    stateCache  *GameStateCache

    // 重连回调
    onReconnect func()
}

// Reconnect 重连
func (rm *ReconnectManager) Reconnect() error {
    for i := 0; i < rm.maxRetries; i++ {
        // 1. 尝试连接
        if err := rm.connect(); err == nil {
            // 连接成功
            rm.connected = true

            // 2. 恢复状态
            if err := rm.restoreState(); err != nil {
                log.Printf("Restore state failed: %v", err)
            }

            // 3. 触发回调
            if rm.onReconnect != nil {
                rm.onReconnect()
            }

            return nil
        }

        // 连接失败，等待重试
        log.Printf("Reconnect failed, retry %d/%d", i+1, rm.maxRetries)
        time.Sleep(rm.retryDelay)
    }

    return errors.New("max retries exceeded")
}

// restoreState 恢复状态
func (rm *ReconnectManager) restoreState() error {
    // 1. 请求状态同步
    state, err := rm.requestStateSync()
    if err != nil {
        return err
    }

    // 2. 应用状态
    rm.applyState(state)

    return nil
}

状态预测

// ClientPrediction 客户端预测
type ClientPrediction struct {
    // 本地状态
    localState  *GameState

    // 服务器状态
    serverState *GameState

    // 预测队列
    predictions []*Prediction
}

type Prediction struct {
    Input       Input
    PredictedState *GameState
}

// OnPlayerInput 玩家输入
func (cp *ClientPrediction) OnPlayerInput(input Input) {
    // 1. 立即预测并显示
    predictedState := cp.localState.Clone()
    predictedState.ApplyInput(input)

    cp.predictions = append(cp.predictions, &Prediction{
        Input:         input,
        PredictedState: predictedState,
    })

    cp.localState = predictedState
    cp.Render(predictedState)

    // 2. 发送输入到服务器
    cp.sendToServer(input)
}

// OnServerCorrection 服务器纠正
func (cp *ClientPrediction) OnServerCorrection(serverState *GameState) {
    // 1. 保存服务器状态
    cp.serverState = serverState

    // 2. 重新应用未确认的输入
    localState := serverState.Clone()

    for _, prediction := range cp.predictions {
        localState.ApplyInput(prediction.Input)
    }

    cp.localState = localState
    cp.Render(localState)

    // 3. 清理已确认的预测
    cp.predictions = cp.predictions[:0]
}

延迟补偿

// LagCompensation 延迟补偿
type LagCompensation struct {
    // 服务器时间
    serverTime  int64

    // 客户端时间差
    timeDiffs  map[uint64]int64  // playerID → timeDiff
}

// Rewind 回溯到客户端看到的时间
func (lc *LagCompensation) Rewind(playerID uint64, clientTime int64) *GameState {
    // 1. 计算时间差
    timeDiff := lc.serverTime - clientTime

    // 2. 获取历史状态
    oldState := lc.stateHistory.GetState(clientTime)

    return oldState
}

// Forward 快进到当前时间
func (lc *LagCompensation) Forward(oldState *GameState, playerID uint64) {
    // 1. 重播未确认的输入
    unackedInputs := lc.getUnackedInputs(playerID)

    for _, input := range unackedInputs {
        oldState.ApplyInput(input)
    }

    // 2. 应用到当前状态
    lc.currentState = oldState
}

国内网络环境

跨地域部署

// GeoDistributedServer 跨地域服务器
type GeoDistributedServer struct {
    // 区域服务器
    regions     map[string]*GameRegion  // "beijing", "shanghai", "guangzhou"

    // 路由策略
    router      *GeoRouter
}

type GameRegion struct {
    Name       string
    Servers    []*GameServer
    Load       float64
}

// SelectRegion 选择区域
func (gds *GeoDistributedServer) SelectRegion(player *Player) *GameRegion {
    // 1. 检查玩家IP归属地
    city := gds.getCityByIP(player.IP)

    // 2. 选择最近的服务器
    switch city {
    case "Beijing", "Tianjin", "Hebei":
        return gds.regions["beijing"]
    case "Shanghai", "Jiangsu", "Zhejiang":
        return gds.regions["shanghai"]
    case "Guangdong", "Guangxi", "Fujian":
        return gds.regions["guangzhou"]
    default:
        // 默认上海
        return gds.regions["shanghai"]
    }
}

// DeployServers 部署服务器
func deployChinaServers() {
    // 华北地区：北京机房
    beijing := &GameRegion{
        Name: "beijing",
        Servers: []*GameServer{
            {IP: "1.2.3.4", Port: 8080},
            {IP: "1.2.3.5", Port: 8080},
        },
    }

    // 华东地区：上海机房
    shanghai := &GameRegion{
        Name: "shanghai",
        Servers: []*GameServer{
            {IP: "5.6.7.8", Port: 8080},
            {IP: "5.6.7.9", Port: 8080},
        },
    }

    // 华南地区：广州机房
    guangzhou := &GameRegion{
        Name: "guangzhou",
        Servers: []*GameServer{
            {IP: "9.10.11.12", Port: 8080},
            {IP: "9.10.11.13", Port: 8080},
        },
    }
}

运营商优化

// ISPOptimizer 运营商优化器
type ISPOptimizer struct {
    // 运营商线路
    lines       map[string]*NetworkLine  // "telecom", "unicom", "mobile"
}

type NetworkLine struct {
    Name    string
    Servers []*GameServer
}

// SelectLine 选择运营商线路
func (io *ISPOptimizer) SelectLine(player *Player) *NetworkLine {
    // 1. 检测玩家运营商
    isp := io.detectISP(player.IP)

    // 2. 选择对应线路
    if line, ok := io.lines[isp]; ok {
        return line
    }

    // 3. 默认电信线路
    return io.lines["telecom"]
}

// detectISP 检测运营商
func (io *ISPOptimizer) detectISP(ip string) string {
    // 简化版：通过IP段判断
    // 实际中需要使用IP数据库

    if strings.HasPrefix(ip, "1.0.") ||
       strings.HasPrefix(ip, "1.2.") {
        return "telecom"  // 电信
    }

    if strings.HasPrefix(ip, "5.0.") ||
       strings.HasPrefix(ip, "5.6.") {
        return "unicom"  // 联通
    }

    if strings.HasPrefix(ip, "9.0.") ||
       strings.HasPrefix(ip, "9.10.") {
        return "mobile"  // 移动
    }

    return "telecom"  // 默认
}

真实案例：《和平精英》弱网优化

背景：

• 目标玩家：移动网络为主
• 弱网环境：地铁、电梯、山区
• 延迟要求：<50ms

技术方案：

1. 协议优化

// 游戏协议优化
func optimizeProtocol() {
    // 1. 使用KCP代替TCP
    // 效果：延迟从120ms降到60ms

    // 2. 优化KCP参数
    sess.SetNoDelay(1, 10, 2, 1)
    // 效果：延迟从60ms降到40ms

    // 3. 启用前向纠错（FEC）
    // 效果：20%丢包下仍可玩
}

2. 弱网适配

// WeakNetworkAdapter 弱网适配器
type WeakNetworkAdapter struct {
    // 网络质量评分
    qualityScore float64
}

// Adapt 网络自适应
func (wna *WeakNetworkAdapter) Adapt() {
    quality := wna.qualityScore

    if quality > 80 {
        // 网络良好
        wna.sendRate = 30  // 30次/秒
        wna.resolution = "1080p"
    } else if quality > 50 {
        // 网络一般
        wna.sendRate = 20  // 20次/秒
        wna.resolution = "720p"
    } else {
        // 网络差
        wna.sendRate = 10  // 10次/秒
        wna.resolution = "480p"
    }
}

效果对比：

小结

房间广播、弱网与国内网络环境的核心要点：

1. 房间广播：批量发送 + 增量更新
2. 弱网优化：断线重连 + 客户端预测 + 延迟补偿
3. 国内环境：跨地域部署 + 运营商优化
4. 网络自适应：根据网络质量调整策略

真实案例：

• 《和平精英》：弱网优化，延迟降低70%
• 《王者荣耀》：跨地域部署，三地机房

踩坑经验：

• ❌ 不要每条消息单独广播
• ❌ 不要忽略移动网络的特殊性
• ✅ 使用客户端预测提升体验

第3章完成！下一章（第4章）我们将学习：同步、战斗与实时交互，深入游戏同步机制。