游戏服务器数据库设计：从玩家数据到排行榜的完整方案

 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

"""
游戏数据分类

玩家数据:
- 账号: 登录, 认证
- 角色: 档案, 属性
- 进度: 等级, 成就

游戏数据:
- 配置: 物品, 任务
- 剧情: 剧情数据
- 地图: 场景数据

运行时数据:
- 状态: 玩家位置, 血量
- 交易: 经济交易
- 聊天: 聊天记录
"""

class GameDataClassification:
    """游戏数据分类"""

    def __init__(self):
        self.categories = {
            "玩家核心数据": {
                "特征": "低频读写",
                "重要性": "高",
                "存储": "关系型数据库",
                "示例": ["账号", "角色档案", "背包"]
            },
            "玩家状态数据": {
                "特征": "高频读写",
                "重要性": "中",
                "存储": "Redis缓存",
                "示例": ["在线状态", "位置", "血量"]
            },
            "游戏配置数据": {
                "特征": "只读",
                "重要性": "高",
                "存储": "文件或NoSQL",
                "示例": ["物品定义", "任务配置"]
            },
            "社交数据": {
                "特征": "关系复杂",
                "重要性": "中",
                "存储": "图数据库或关系型",
                "示例": ["好友", "公会", "聊天"]
            },
            "日志数据": {
                "特征": "海量只写",
                "重要性": "低",
                "存储": "日志系统",
                "示例": ["登录日志", "行为日志"]
            }
        }

    def storage_strategy(self):
        """存储策略"""

        strategy = {
            "关系型数据库": {
                "适用": "结构化数据, 事务",
                "技术": ["MySQL", "PostgreSQL"],
                "优化": "索引, 分区"
            },
            "NoSQL": {
                "适用": "灵活schema, 大数据",
                "技术": ["MongoDB", "Cassandra"],
                "优化": "数据模型, 分片"
            },
            "缓存": {
                "适用": "热数据, 计数",
                "技术": ["Redis", "Memcached"],
                "优化": "缓存策略, 过期"
            },
            "搜索引擎": {
                "适用": "全文搜索",
                "技术": ["Elasticsearch"],
                "应用": "玩家搜索, 日志分析"
            }
        }

        return strategy

 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

class PlayerDataModel:
    """玩家数据模型"""

    def __init__(self):
        self.tables = {
            "账号表": {
                "user_id": "PK",
                "username": "UK",
                "email": "UK",
                "password_hash": "",
                "created_at": "",
                "last_login": "",
                "索引": ["username", "email"]
            },
            "角色表": {
                "character_id": "PK",
                "user_id": "FK",
                "name": "",
                "level": "",
                "class": "",
                "exp": "",
                "gold": "",
                "last_save": "",
                "索引": ["user_id", "name"]
            },
            "背包表": {
                "item_id": "PK",
                "character_id": "FK",
                "item_template_id": "",
                "quantity": "",
                "slot": "",
                "enchant": "",
                "索引": ["character_id"]
            }
        }

    def relationship_design(self):
        """关系设计"""

        relationships = {
            "一对多": {
                "用户→角色": "一个用户多个角色",
                "角色→物品": "一个角色多个物品",
                "实现": "外键约束"
            },
            "多对多": {
                "好友关系": "好友表",
                "公会成员": "公会成员表",
                "实现": "中间表"
            },
            "一对多自引用": {
                "任务链": "前置任务",
                "技能树": "前置技能",
                "实现": "自引用外键"
            }
        }

        return relationships

 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

class ShardingStrategy:
    """分库分表策略"""

    def __init__(self):
        self.methods = {
            "水平分表": {
                "按玩家ID": "范围或哈希",
                "按时间": "时间范围",
                "按区域": "游戏区域"
            },
            "垂直分库": {
                "按业务": "用户库, 游戏库",
                "按读写": "主库, 从库",
                "按数据类型": "关系型, NoSQL"
            }
        }

    def sharding_approach(self):
        """分片方法"""

        approach = {
            "范围分片": {
                "方法": "按ID范围",
                "优势": "简单查询",
                "劣势": "热点问题",
                "示例": "0-100万, 100-200万"
            },
            "哈希分片": {
                "方法": "哈希函数",
                "优势": "分布均匀",
                "劣势": "跨片查询",
                "示例": "id % 4"
            },
            "地理位置": {
                "方法": "按玩家位置",
                "优势": "就近访问",
                "应用": "多地域部署"
            }
        }

        return approach

    def distributed_id(self):
        """分布式ID"""

        id_generation = {
            "雪花算法": {
                "结构": "时间戳+机器ID+序列",
                "优势": "有序, 唯一",
                "实现": "简单"
            },
            "UUID": {
                "类型": "UUID v4",
                "优势": "简单",
                "劣势": "无序, 较长"
            },
            "自增": {
                "问题": "分布式困难",
                "方案": "号段模式",
                "应用": "单机"
            }
        }

        return id_generation

 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

class CacheArchitecture:
    """缓存架构"""

    def __init__(self):
        self.layers = {
            "本地缓存": {
                "位置": "应用进程内存",
                "容量": "小(MB级)",
                "速度": "最快",
                "技术": ["Guava", "Caffeine"]
            },
            "分布式缓存": {
                "位置": "独立缓存集群",
                "容量": "大(GB级)",
                "速度": "快(网络)",
                "技术": ["Redis", "Memcached"]
            },
            "CDN缓存": {
                "位置": "边缘节点",
                "内容": "静态资源",
                "技术": ["Cloudflare", "Akamai"]
            }
        }

    def redis_patterns(self):
        """Redis模式"""

        patterns = {
            "缓存穿透": {
                "问题": "查询不存在的数据",
                "解决": "布隆过滤器",
                "空值缓存": "缓存空结果"
            },
            "缓存击穿": {
                "问题": "热点key过期",
                "解决": "互斥锁",
                "永不过期": "逻辑过期"
            },
            "缓存雪崩": {
                "问题": "大量key同时过期",
                "解决": "随机过期时间",
                "熔断": "限流降级"
            }
        }

        return patterns

 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

class LeaderboardImplementation:
    """排行榜实现"""

    def __init__(self):
        self.requirements = {
            "实时性": {
                "更新": "即时更新",
                "查询": "快速查询",
                "排名": "实时排名"
            },
            "性能": {
                "并发": "高并发读写",
                "延迟": "低延迟响应",
                "规模": "百万级玩家"
            }
        }

    def redis_sorted_set(self):
        """Redis Sorted Set实现"""

        implementation = {
            "数据结构": {
                "类型": "Sorted Set",
                "成员": "玩家ID",
                "分数": "分数/积分"
            },
            "操作": {
                "ZADD": "添加或更新分数",
                "ZINCRBY": "增加分数",
                "ZRANGE": "获取排名",
                "ZREVRANK": "获取排名"
            },
            "优化": {
                "分片": "按范围分片",
                "过期": "定期清理",
                "快照": "定期快照"
            }
        }

        return implementation

    def advanced_features(self):
        """高级特性"""

        features = {
            "多维度": {
                "全服": "全服排行",
                "好友": "好友排行",
                "公会": "公会排行"
            },
            "时效性": {
                "实时": "实时更新",
                "每日": "每日重置",
                "每周": "每周排行"
            },
            "奖励": {
                "定时": "根据排名发奖",
                "通知": "结果通知",
                "记录": "历史记录"
            }
        }

        return features

 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

class DataConsistency:
    """数据一致性"""

    def __init__(self):
        self.scenarios = {
            "交易": {
                "需求": "强一致性",
                "方法": "数据库事务",
                "隔离": "Serializable隔离"
            },
            "排行榜": {
                "需求": "最终一致性",
                "方法": "Redis + 异步持久化",
                "优化": "定期同步"
            },
            "社交": {
                "需求": "最终一致性",
                "方法": "消息队列",
                "重试": "失败重试"
            }
        }

    def transaction_patterns(self):
        """事务模式"""

        patterns = {
            "本地事务": {
                "描述": "单数据库事务",
                "实现": "BEGIN/COMMIT",
                "应用": "单库操作"
            },
            "分布式事务": {
                "2PC": "两阶段提交",
                "Saga": "补偿事务",
                "TCC": "Try-Confirm-Cancel"
            },
            "最终一致性": {
                "事件": "事件驱动",
                "重试": "指数退避",
                "死信队列": "失败处理"
            }
        }

        return patterns

 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

class DatabaseOptimization:
    """数据库优化"""

    def __init__(self):
        self.techniques = {
            "索引": {
                "主键索引": "自动创建",
                "唯一索引": "唯一约束",
                "复合索引": "多列索引",
                "覆盖索引": "包含查询列"
            },
            "查询": {
                "优化": "EXPLAIN分析",
                "避免": "SELECT *",
                "限制": "LIMIT分页",
                "缓存": "查询结果缓存"
            },
            "表设计": {
                "范式": "适当范式化",
                "反范式": "查询优化",
                "分区": "表分区",
                "分表": "水平拆分"
            }
        }

    def monitoring(self):
        """监控指标"""

        metrics = {
            "性能": {
                "QPS": "每秒查询数",
                "延迟": "查询响应时间",
                "慢查询": "慢查询日志"
            },
            "资源": {
                "连接": "连接池使用",
                "CPU": "数据库CPU",
                "内存": "缓冲池使用",
                "磁盘": "IO使用"
            },
            "复制": {
                "延迟": "主从延迟",
                "状态": "复制状态",
                "可用": "从库可用性"
            }
        }

        return metrics

 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

class BestPractices:
    """最佳实践"""

    def __init__(self):
        self.principles = {
            "设计": {
                "简单": "开始简单",
                "范式": "适当范式化",
                "索引": "合理索引",
                "分区": "提前规划"
            },
            "开发": {
                "迁移": "版本迁移",
                "测试": "充分测试",
                "备份": "定期备份",
                "监控": "持续监控"
            },
            "运维": {
                "高可用": "主从复制",
                "读写分离": "分离读写",
                "负载均衡": "均衡负载",
                "故障转移": "自动转移"
            }
        }

    def common_mistakes(self):
        """常见错误"""

        mistakes = {
            "过度索引": {
                "问题": "索引太多",
                "影响": "写入性能",
                "解决": "只索引需要的列"
            },
            "N+1查询": {
                "问题": "循环查询",
                "解决": "JOIN或批量查询",
                "示例": "批量加载玩家数据"
            },
            "大事务": {
                "问题": "事务太长",
                "影响": "锁定, 死锁",
                "解决": "拆分事务"
            }
        }

        return mistakes