引言

在现代在线游戏开发中，服务器与客户端之间实时、可靠的通信机制是游戏体验的基石。作为一名游戏后端开发者，我曾经遇到过这样的场景：更新了一个公会系统的新功能，服务器需要向成千上万个在线玩家推送公会状态变更。短短几小时后，服务器内存使用率飙升至 90%，系统告警不断。问题出在哪里？Push 消息的可靠性机制实现不当导致了内存泄漏。

本文将深入探讨游戏后端中 Push-ACK 机制的设计与实现，特别关注如何避免内存暴涨问题，分享我在多个大型游戏项目中积累的经验与教训。

背景：为什么需要应用层的 ACK 机制？

TCP 协议确实提供了可靠的数据传输保证，包括数据包的序列号、校验和、超时重传等机制。那么，为什么我们还需要在应用层实现额外的 ACK 机制呢？

TCP 可靠性的边界

TCP 只能保证数据被送达到客户端的网络栈，但无法保证：

  1. 数据被客户端应用程序正确处理
  2. 处理过程中没有出现异常
  3. 客户端的业务逻辑正确执行

想象这样一个场景：服务器向玩家推送了一条"获得稀有装备"的消息，TCP 确保了数据送达客户端，但如果客户端在处理这个消息时崩溃了呢？对于游戏这类状态敏感的应用，我们需要知道消息是否被成功处理，而不仅仅是成功传输。

业务可靠性需求

实际游戏开发中，不同类型的消息有不同的可靠性需求：

消息类型 示例 可靠性需求
关键状态变更 道具获取、货币变化 极高（必须确认处理）
游戏进程通知 任务更新、成就解锁 高（需要确认）
实时位置同步 玩家位置、NPC 移动 中（新数据可覆盖旧数据）
环境信息 天气变化、背景音乐 低（可接受偶尔丢失）

设计通用的 Push-ACK 机制

一个完善的 Push-ACK 机制需要考虑以下几个方面：消息唯一标识、优先级分级、超时重试、批量确认和失败处理。下面是基于 Go 语言的设计实现：

核心数据结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
// Message 表示服务器推送的消息
type Message struct {
    MsgID        string      `json:"msg_id"`        // 唯一消息标识
    MsgType      string      `json:"msg_type"`      // 消息类型
    Timestamp    int64       `json:"timestamp"`     // 发送时间戳
    Priority     int         `json:"priority"`      // 优先级：1-高，2-中，3-低
    Payload      interface{} `json:"payload"`       // 消息内容
    RequiresAck  bool        `json:"requires_ack"`  // 是否需要确认
    Expiration   int64       `json:"expiration"`    // 过期时间戳
}

// AckMessage 表示客户端的确认消息
type AckMessage struct {
    AckID            string  `json:"ack_id"`           // 对应原消息ID
    Status           string  `json:"status"`           // 状态：success/failed/partial
    ClientTimestamp  int64   `json:"client_timestamp"` // 客户端处理时间
    ErrorCode        int     `json:"error_code"`       // 错误码
    ErrorMessage     string  `json:"error_message"`    // 错误信息
}

// BatchAckMessage 表示批量确认消息
type BatchAckMessage struct {
    BatchAck        bool     `json:"batch_ack"`       // 批量确认标志
    AckIDs          []string `json:"ack_ids"`         // 消息ID列表
    Status          string   `json:"status"`          // 状态
    ClientTimestamp int64    `json:"client_timestamp"`// 确认时间
}

// PendingMessageInfo 表示等待确认的消息信息
type PendingMessageInfo struct {
    ClientID    string      // 客户端ID
    Message     *Message    // 原始消息
    SentTime    int64       // 发送时间
    RetryCount  int         // 重试次数
}

服务器端 Push 管理器实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
// PushManager 负责管理推送消息和确认
type PushManager struct {
    pendingMessages    map[string]*PendingMessageInfo  // 等待确认的消息
    clientMessageCount map[string]int                  // 每个客户端的消息数量
    ackTimeout         int64                           // 确认超时时间(秒)
    maxRetries         int                             // 最大重试次数
    maxPendingPerClient int                            // 每客户端最大消息数
    maxMessageAge      int64                           // 消息最大生存时间(秒)

    // 内存监控相关
    memoryThresholdMB  int64                           // 内存阈值(MB)
    criticalThresholdMB int64                          // 危险内存阈值(MB)

    mutex              sync.RWMutex                    // 保护并发访问

    // 网络接口（依赖外部实现）
    networkLayer       NetworkInterface
}

// NewPushManager 创建一个新的推送管理器
func NewPushManager(networkLayer NetworkInterface) *PushManager {
    pm := &PushManager{
        pendingMessages:     make(map[string]*PendingMessageInfo),
        clientMessageCount:  make(map[string]int),
        ackTimeout:          10,
        maxRetries:          3,
        maxPendingPerClient: 1000,
        maxMessageAge:       300,
        memoryThresholdMB:   1000,  // 1GB
        criticalThresholdMB: 1500,  // 1.5GB
        networkLayer:        networkLayer,
    }

    // 启动后台任务
    go pm.checkTimeoutsLoop()
    go pm.cleanupLoop()
    go pm.memoryMonitorLoop()

    return pm
}

// PushMessage 向客户端推送消息
func (pm *PushManager) PushMessage(clientID string, message *Message) bool {
    // 如果不需要确认，直接发送
    if !message.RequiresAck {
        return pm.networkLayer.SendToClient(clientID, message)
    }

    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    // 检查客户端消息数是否超限
    if pm.clientMessageCount[clientID] >= pm.maxPendingPerClient {
        pm.handleQueueOverflow(clientID, message)
        return false
    }

    // 存储待确认消息
    pm.pendingMessages[message.MsgID] = &PendingMessageInfo{
        ClientID:    clientID,
        Message:     message,
        SentTime:    time.Now().Unix(),
        RetryCount:  0,
    }

    // 更新客户端消息计数
    pm.clientMessageCount[clientID]++

    // 发送消息
    return pm.networkLayer.SendToClient(clientID, message)
}

// ProcessAck 处理客户端的确认消息
func (pm *PushManager) ProcessAck(clientID string, ack *AckMessage) bool {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    info, exists := pm.pendingMessages[ack.AckID]
    if !exists || info.ClientID != clientID {
        return false
    }

    // 确认成功，删除消息
    delete(pm.pendingMessages, ack.AckID)
    pm.clientMessageCount[clientID]--

    // 如果客户端没有待确认消息了，清理计数器
    if pm.clientMessageCount[clientID] <= 0 {
        delete(pm.clientMessageCount, clientID)
    }

    return true
}

// ProcessBatchAck 处理批量确认
func (pm *PushManager) ProcessBatchAck(clientID string, batchAck *BatchAckMessage) int {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    confirmedCount := 0

    for _, ackID := range batchAck.AckIDs {
        info, exists := pm.pendingMessages[ackID]
        if exists && info.ClientID == clientID {
            delete(pm.pendingMessages, ackID)
            pm.clientMessageCount[clientID]--
            confirmedCount++
        }
    }

    // 如果客户端没有待确认消息了，清理计数器
    if pm.clientMessageCount[clientID] <= 0 {
        delete(pm.clientMessageCount, clientID)
    }

    return confirmedCount
}

// 后台任务：超时检查与重试
func (pm *PushManager) checkTimeoutsLoop() {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        pm.checkTimeouts()
    }
}

// 超时检查与重试
func (pm *PushManager) checkTimeouts() {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    now := time.Now().Unix()

    for msgID, info := range pm.pendingMessages {
        // 检查是否超时
        if now - info.SentTime > pm.ackTimeout {
            if info.RetryCount < pm.maxRetries {
                // 增加重试次数
                info.RetryCount++
                info.SentTime = now

                // 重新发送
                pm.networkLayer.SendToClient(info.ClientID, info.Message)
                log.Printf("Retrying message %s to client %s, attempt %d",
                          msgID, info.ClientID, info.RetryCount)
            } else {
                // 超出最大重试次数，放弃并记录
                log.Printf("Message %s to client %s failed after %d attempts",
                          msgID, info.ClientID, pm.maxRetries)

                delete(pm.pendingMessages, msgID)
                pm.clientMessageCount[info.ClientID]--

                // 通知业务层处理失败
                go pm.notifyMessageFailed(info.ClientID, info.Message)
            }
        }
    }
}

解决内存暴涨问题

在大型游戏中，服务器可能同时维护数十万甚至上百万个连接，如果每个连接都有数百条待确认消息，服务器内存很快就会爆满。以下是我在实践中总结的几种高效内存管理策略：

1. 周期性过期消息清理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
// 清理过期消息的后台循环
func (pm *PushManager) cleanupLoop() {
    ticker := time.NewTicker(1 * time.Minute)
    defer ticker.Stop()

    for range ticker.C {
        pm.cleanExpiredMessages()
    }
}

// 清理过期消息
func (pm *PushManager) cleanExpiredMessages() {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    now := time.Now().Unix()
    expiredCount := 0

    for msgID, info := range pm.pendingMessages {
        // 检查消息是否过期
        if now - info.SentTime > pm.maxMessageAge {
            delete(pm.pendingMessages, msgID)
            pm.clientMessageCount[info.ClientID]--
            expiredCount++

            // 记录日志
            log.Printf("Cleaned expired message %s to client %s (age: %d seconds)",
                      msgID, info.ClientID, now - info.SentTime)
        }
    }

    if expiredCount > 0 {
        log.Printf("Cleanup: Removed %d expired messages", expiredCount)
    }
}

2. 消息压缩与合并

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
// CompressMessage 压缩消息以减少内存占用
func CompressMessage(message *Message) []byte {
    // 将消息转为JSON
    jsonData, err := json.Marshal(message)
    if err != nil {
        log.Printf("Error marshaling message: %v", err)
        return nil
    }

    // 使用gzip压缩
    var buf bytes.Buffer
    writer := gzip.NewWriter(&buf)

    _, err = writer.Write(jsonData)
    if err != nil {
        log.Printf("Error compressing message: %v", err)
        return nil
    }

    if err := writer.Close(); err != nil {
        log.Printf("Error closing gzip writer: %v", err)
        return nil
    }

    return buf.Bytes()
}

// DecompressMessage 解压缩消息
func DecompressMessage(compressed []byte) (*Message, error) {
    reader, err := gzip.NewReader(bytes.NewReader(compressed))
    if err != nil {
        return nil, fmt.Errorf("create gzip reader: %w", err)
    }
    defer reader.Close()

    var buf bytes.Buffer
    if _, err := io.Copy(&buf, reader); err != nil {
        return nil, fmt.Errorf("decompress data: %w", err)
    }

    var message Message
    if err := json.Unmarshal(buf.Bytes(), &message); err != nil {
        return nil, fmt.Errorf("unmarshal json: %w", err)
    }

    return &message, nil
}

3. 分级存储策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
// PushManager 增加分级存储功能
type PushManager struct {
    // ... 之前的字段 ...

    // 内存中存储高优先级消息
    memoryPending     map[string]*PendingMessageInfo

    // Redis客户端，用于存储低优先级消息
    redisClient      *redis.Client
    redisKeyPrefix   string
    redisExpiry      time.Duration
}

// PushMessage 分级存储版本
func (pm *PushManager) PushMessage(clientID string, message *Message) bool {
    // 如果不需要确认，直接发送
    if !message.RequiresAck {
        return pm.networkLayer.SendToClient(clientID, message)
    }

    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    // 检查客户端消息数量限制
    if pm.clientMessageCount[clientID] >= pm.maxPendingPerClient {
        pm.handleQueueOverflow(clientID, message)
        return false
    }

    pm.clientMessageCount[clientID]++

    // 根据优先级选择存储位置
    if message.Priority <= 2 { // 高优先级和中优先级
        // 存入内存
        pm.memoryPending[message.MsgID] = &PendingMessageInfo{
            ClientID:    clientID,
            Message:     message,
            SentTime:    time.Now().Unix(),
            RetryCount:  0,
        }
    } else { // 低优先级
        // 存入Redis
        messageInfo := &PendingMessageInfo{
            ClientID:    clientID,
            Message:     message,
            SentTime:    time.Now().Unix(),
            RetryCount:  0,
        }

        jsonData, err := json.Marshal(messageInfo)
        if err != nil {
            log.Printf("Error marshaling message: %v", err)
            pm.clientMessageCount[clientID]--
            return false
        }

        redisKey := pm.redisKeyPrefix + message.MsgID
        err = pm.redisClient.Set(context.Background(), redisKey, jsonData, pm.redisExpiry).Err()
        if err != nil {
            log.Printf("Error storing message in Redis: %v", err)
            pm.clientMessageCount[clientID]--
            return false
        }
    }

    // 发送消息
    return pm.networkLayer.SendToClient(clientID, message)
}

// ProcessAck 分级存储版本
func (pm *PushManager) ProcessAck(clientID string, ack *AckMessage) bool {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    // 先检查内存中的消息
    info, existsInMemory := pm.memoryPending[ack.AckID]
    if existsInMemory && info.ClientID == clientID {
        delete(pm.memoryPending, ack.AckID)
        pm.clientMessageCount[clientID]--

        if pm.clientMessageCount[clientID] <= 0 {
            delete(pm.clientMessageCount, clientID)
        }

        return true
    }

    // 再检查Redis中的消息
    redisKey := pm.redisKeyPrefix + ack.AckID
    exists, err := pm.redisClient.Exists(context.Background(), redisKey).Result()
    if err != nil {
        log.Printf("Error checking message in Redis: %v", err)
        return false
    }

    if exists == 1 {
        // 获取消息以验证客户端ID
        jsonData, err := pm.redisClient.Get(context.Background(), redisKey).Bytes()
        if err != nil {
            log.Printf("Error getting message from Redis: %v", err)
            return false
        }

        var messageInfo PendingMessageInfo
        if err := json.Unmarshal(jsonData, &messageInfo); err != nil {
            log.Printf("Error unmarshaling message from Redis: %v", err)
            return false
        }

        if messageInfo.ClientID == clientID {
            // 从Redis删除并更新计数
            pm.redisClient.Del(context.Background(), redisKey)
            pm.clientMessageCount[clientID]--

            if pm.clientMessageCount[clientID] <= 0 {
                delete(pm.clientMessageCount, clientID)
            }

            return true
        }
    }

    return false
}

4. 内存自适应调整

内存自适应调整是我在实际项目中解决突发流量问题的关键策略。它能够根据当前系统负载动态调整消息处理参数，确保系统稳定性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
// 内存监控循环
func (pm *PushManager) memoryMonitorLoop() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        memoryMB := pm.getMemoryUsageMB()

        if memoryMB > pm.criticalThresholdMB {
            // 紧急情况，进行应急清理
            pm.emergencyCleanup(memoryMB)
        } else if memoryMB > pm.memoryThresholdMB {
            // 超过警戒线，调整参数
            pm.adjustParameters(memoryMB)
        }
    }
}

// 获取当前进程内存使用量（MB）
func (pm *PushManager) getMemoryUsageMB() int64 {
    var memStats runtime.MemStats
    runtime.ReadMemStats(&memStats)
    return int64(memStats.Alloc / 1024 / 1024)
}

// 根据内存使用情况调整参数
func (pm *PushManager) adjustParameters(currentMemoryMB int64) {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    // 计算内存超出比例
    excessRatio := float64(currentMemoryMB - pm.memoryThresholdMB) / float64(pm.memoryThresholdMB)

    // 调整每客户端最大消息数
    newMaxPerClient := int(float64(pm.maxPendingPerClient) * (1 - excessRatio*0.5))
    if newMaxPerClient < 100 {
        newMaxPerClient = 100 // 确保至少保留100条
    }

    // 调整消息最大生存时间
    newMaxAge := int64(float64(pm.maxMessageAge) * (1 - excessRatio*0.5))
    if newMaxAge < 60 {
        newMaxAge = 60 // 至少60秒
    }

    // 更新参数
    pm.maxPendingPerClient = newMaxPerClient
    pm.maxMessageAge = newMaxAge

    log.Printf("Memory usage: %d MB, adjusted parameters: maxPending=%d, maxAge=%ds",
               currentMemoryMB, pm.maxPendingPerClient, pm.maxMessageAge)

    // 执行一次清理
    pm.cleanExpiredMessages()
}

// 紧急清理
func (pm *PushManager) emergencyCleanup(currentMemoryMB int64) {
    pm.mutex.Lock()
    defer pm.mutex.Unlock()

    log.Printf("CRITICAL: Memory usage at %d MB, performing emergency cleanup", currentMemoryMB)

    // 大幅降低参数
    pm.maxPendingPerClient = 100
    pm.maxMessageAge = 60

    // 清理低优先级消息
    for msgID, info := range pm.memoryPending {
        if info.Message.Priority > 1 { // 只保留最高优先级
            delete(pm.memoryPending, msgID)
            pm.clientMessageCount[info.ClientID]--
        }
    }

    log.Printf("Emergency cleanup completed")
}

5. 队列溢出处理策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
// 处理队列溢出
func (pm *PushManager) handleQueueOverflow(clientID string, newMessage *Message) {
    log.Printf("Queue overflow for client %s", clientID)

    // 策略1: 根据消息优先级决定是否替换现有消息
    if newMessage.Priority == 1 { // 高优先级消息
        // 查找并替换该客户端的一条低优先级消息
        for msgID, info := range pm.memoryPending {
            if info.ClientID == clientID && info.Message.Priority > 1 {
                // 记录
                log.Printf("Replacing low priority message %s with high priority message", msgID)

                // 删除旧消息
                delete(pm.memoryPending, msgID)

                // 添加新消息
                pm.memoryPending[newMessage.MsgID] = &PendingMessageInfo{
                    ClientID:    clientID,
                    Message:     newMessage,
                    SentTime:    time.Now().Unix(),
                    RetryCount:  0,
                }

                // 发送新消息
                pm.networkLayer.SendToClient(clientID, newMessage)
                return
            }
        }
    }

    // 策略2: 丢弃旧消息以腾出空间
    // 查找该客户端最旧的消息
    var oldestMsgID string
    var oldestTime int64 = math.MaxInt64

    for msgID, info := range pm.memoryPending {
        if info.ClientID == clientID && info.SentTime < oldestTime {
            oldestMsgID = msgID
            oldestTime = info.SentTime
        }
    }

    if oldestMsgID != "" {
        log.Printf("Dropping oldest message %s for client %s", oldestMsgID, clientID)
        delete(pm.memoryPending, oldestMsgID)

        // 添加新消息
        pm.memoryPending[newMessage.MsgID] = &PendingMessageInfo{
            ClientID:    clientID,
            Message:     newMessage,
            SentTime:    time.Now().Unix(),
            RetryCount:  0,
        }

        // 发送新消息
        pm.networkLayer.SendToClient(clientID, newMessage)
    } else {
        // 极端情况，无法找到可替换的消息
        log.Printf("Cannot find message to replace for client %s", clientID)
    }
}

客户端实现

客户端实现同样关键，特别是批量确认机制能显著减少网络流量：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
// PushReceiver 客户端推送接收处理器
type PushReceiver struct {
    connection        Connection          // 网络连接接口
    processedMsgIDs   map[string]int64    // 已处理消息ID及处理时间
    pendingAcks       []string            // 待确认的消息ID
    ackBatchSize      int                 // 批量确认大小
    ackInterval       time.Duration       // 批量确认间隔
    messageHandlers   map[string]MessageHandler // 消息处理函数

    mutex             sync.Mutex          // 保护并发访问
    stopChan          chan struct{}       // 停止信号
}

// MessageHandler 消息处理函数类型
type MessageHandler func(payload interface{}) error

// NewPushReceiver 创建推送接收器
func NewPushReceiver(conn Connection) *PushReceiver {
    receiver := &PushReceiver{
        connection:       conn,
        processedMsgIDs:  make(map[string]int64),
        pendingAcks:      make([]string, 0, 100),
        ackBatchSize:     50,
        ackInterval:      time.Second,
        messageHandlers:  make(map[string]MessageHandler),
        stopChan:         make(chan struct{}),
    }

    // 启动批量确认任务
    go receiver.ackLoop()

    // 启动过期消息ID清理任务
    go receiver.cleanupLoop()

    return receiver
}

// RegisterHandler 注册消息处理函数
func (r *PushReceiver) RegisterHandler(msgType string, handler MessageHandler) {
    r.mutex.Lock()
    defer r.mutex.Unlock()

    r.messageHandlers[msgType] = handler
}

// HandleMessage 处理收到的消息
func (r *PushReceiver) HandleMessage(message *Message) {
    r.mutex.Lock()
    defer r.mutex.Unlock()

    msgID := message.MsgID

    // 检查是否已处理过该消息
    if _, exists := r.processedMsgIDs[msgID]; exists {
        // 已处理过，再次发送确认
        if message.RequiresAck {
            r.pendingAcks = append(r.pendingAcks, msgID)

            // 如果积累的确认数量超过批量大小，立即发送
            if len(r.pendingAcks) >= r.ackBatchSize {
                go r.sendBatchAcks()
            }
        }
        return
    }

    // 查找处理函数
    handler, exists := r.messageHandlers[message.MsgType]
    if !exists {
        log.Printf("No handler for message type: %s", message.MsgType)

        // 未知消息类型也需要确认
        if message.RequiresAck {
            r.sendErrorAck(msgID, "Unknown message type")
        }
        return
    }

    // 处理消息
    err := handler(message.Payload)
    if err != nil {
        log.Printf("Error processing message %s: %v", msgID, err)

        if message.RequiresAck {
            r.sendErrorAck(msgID, err.Error())
        }
        return
    }

    // 记录已处理的消息
    r.processedMsgIDs[msgID] = time.Now().Unix()

    // 如果需要确认，加入待确认队列
    if message.RequiresAck {
        r.pendingAcks = append(r.pendingAcks, msgID)

        // 如果积累的确认数量超过批量大小，立即发送
        if len(r.pendingAcks) >= r.ackBatchSize {
            go r.sendBatchAcks()
        }
    }
}

// 发送批量确认
func (r *PushReceiver) sendBatchAcks() {
    r.mutex.Lock()

    // 如果没有待确认消息，直接返回
    if len(r.pendingAcks) == 0 {
        r.mutex.Unlock()
        return
    }

    // 复制当前的待确认ID列表
    ackIDs := make([]string, len(r.pendingAcks))
    copy(ackIDs, r.pendingAcks)

    // 清空待确认列表
    r.pendingAcks = r.pendingAcks[:0]

    r.mutex.Unlock()

    // 创建批量确认消息
    batchAck := &BatchAckMessage{
        BatchAck:        true,
        AckIDs:          ackIDs,
        Status:          "success",
        ClientTimestamp: time.Now().Unix(),
    }

    // 发送确认
    r.connection.Send(batchAck)
}

// 发送错误确认
func (r *PushReceiver) sendErrorAck(msgID string, errorMessage string) {
    ack := &AckMessage{
        AckID:           msgID,
        Status:          "failed",
        ClientTimestamp: time.Now().Unix(),
        ErrorCode:       1001,
        ErrorMessage:    errorMessage,
    }

    r.connection.Send(ack)
}

// 批量确认定时器
func (r *PushReceiver) ackLoop() {
    ticker := time.NewTicker(r.ackInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            r.sendBatchAcks()
        case <-r.stopChan:
            return
        }
    }
}

// 清理过期的已处理消息ID
func (r *PushReceiver) cleanupLoop() {
    // 每小时清理一次
    ticker := time.NewTicker(1 * time.Hour)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            r.cleanupProcessedIDs()
        case <-r.stopChan:
            return
        }
    }
}

// 清理过期的已处理消息ID
func (r *PushReceiver) cleanupProcessedIDs() {
    r.mutex.Lock()
    defer r.mutex.Unlock()

    now := time.Now().Unix()
    expireTime := int64(86400) // 24小时过期

    for msgID, processTime := range r.processedMsgIDs {
        if now - processTime > expireTime {
            delete(r.processedMsgIDs, msgID)
        }
    }
}

// Close 关闭推送接收器
func (r *PushReceiver) Close() {
    // 发送所有待确认消息
    r.sendBatchAcks()

    // 停止所有后台任务
    close(r.stopChan)
}

实战经验与最佳实践

在多个千万用户级别的游戏项目实践中，我总结了以下几点 Push-ACK 机制的最佳实践：

1. 消息分级是关键

不是所有消息都需要相同级别的可靠性保证。在一个 MMORPG 项目中，我们将消息分为四级：

  • 关键级：直接影响游戏平衡和经济的消息，如道具获取、货币变化
  • 重要级：影响游戏进程的消息，如任务更新、排行榜变动
  • 普通级：一般游戏状态信息，如其他玩家动作、环境变化
  • 低优先级：可以容忍丢失的背景信息，如聊天、天气效果

高级别消息使用完整的 ACK 机制，低级别消息可以简化甚至取消 ACK 需求，这样大大减轻了服务器内存压力。

2. 利用统计指标进行调优

监控以下关键指标：

  • ACK 响应时间分布
  • 消息重试率
  • 每客户端平均待确认消息数
  • 内存使用增长曲线

在一个足球经理类游戏中，通过这些指标我们发现，将 ACK 超时时间从 10 秒调整到 5 秒，并将最大重试次数从 3 次增加到 5 次，可以将消息最终确认率从 99.2%提高到 99.8%，同时减少了 25%的内存使用。

3. 针对不同网络环境优化

移动网络环境差异很大，针对不同网络条件动态调整策略：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 根据网络条件调整参数
func (pm *PushManager) adjustForNetworkCondition(clientID string, rtt time.Duration) {
    // 网络条件良好
    if rtt < 100*time.Millisecond {
        pm.clientTimeouts[clientID] = 3 // 3秒超时
        pm.clientRetries[clientID] = 2  // 2次重试
    } else if rtt < 300*time.Millisecond {
        pm.clientTimeouts[clientID] = 5 // 5秒超时
        pm.clientRetries[clientID] = 3  // 3次重试
    } else {
        pm.clientTimeouts[clientID] = 10 // 10秒超时
        pm.clientRetries[clientID] = 5   // 5次重试
    }
}

4. 定期压力测试

在一个大型开放世界游戏中，我们每月进行一次"混沌测试"，模拟极端情况：

  1. 突发 50%客户端同时掉线然后重连
  2. 模拟网络延迟突然从 50ms 增加到 500ms
  3. 模拟 10%的确认消息丢失

这种测试让我们发现了很多边缘情况，并建立了更健壮的防御机制。

结论

一个设计良好的 Push-ACK 机制是现代游戏服务器架构的核心组件。它确保了游戏状态的一致性，提升了玩家体验，同时也为运营团队提供了可靠的数据基础。最重要的是，它必须是高性能且资源友好的。

通过采用本文介绍的多级存储、自适应参数调整、消息优先级和过期策略等技术，我们可以构建一个既可靠又高效的推送确认系统，即使在面对数十万并发

许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。

分享文章

快来参与讨论吧~