第七部分：NoSQL 数据库与混合持久化

关系型数据库并非万能，在特定场景下 NoSQL 数据库能提供更好的性能和灵活性。进阶开发者应当理解不同类型数据库的适用场景，并根据业务需求组合使用（混合持久化架构，Polyglot Persistence）。

7.1 常见 NoSQL 分类与选型

7.2 Redis 高级用法与持久化
Redis 不仅是缓存，还可以作为主数据库或消息队列。

7.2.1 数据结构的高级使用

# 哈希表存储对象（优于多 key）
HSET user:1001 name "Alice" age 30
HGETALL user:1001

# 有序集合做排行榜
ZADD leaderboard 100 "playerA" 95 "playerB"
ZREVRANGE leaderboard 0 2 WITHSCORES

# 位图统计日活（每个用户占用1位）
SETBIT login:2025-06-01 1001 1
BITCOUNT login:2025-06-01

7.2.2 持久化机制
RDB：快照方式，定时生成全量备份。恢复快但可能丢失最近一次快照后的数据。

AOF：追加写命令日志，更安全（可配置每秒 fsync），但文件体积大、恢复慢。

生产环境通常两者同时开启，或者使用 Redis Enterprise 的持久化方案。

7.2.3 高可用与集群
Redis Sentinel：提供主从自动切换，实现高可用。

Redis Cluster：分片方案，支持横向扩展，但部分跨 key 操作受限。

7.3 MongoDB 的索引与聚合
MongoDB 的索引机制类似关系数据库，但支持嵌套字段索引、地理空间索引、全文索引等。

聚合框架：替代复杂的分组和连接，通过管道（$match, $group, $lookup）处理数据。例如，统计每个商品的销量前 10 的用户：

db.orders.aggregate([
    { $match: { status: "completed" } },
    { $group: { _id: { productId: "$product_id", userId: "$user_id" }, total: { $sum: "$amount" } } },
    { $sort: { total: -1 } },
    { $group: { _id: "$_id.productId", topUsers: { $push: { userId: "$_id.userId", total: "$total" } } } },
    { $project: { topUsers: { $slice: ["$topUsers", 10] } } }
]);

7.4 混合持久化架构示例
一个典型的电商系统可能同时使用：

MySQL：存储核心订单、用户账户（强事务、强一致性）。

Redis：缓存商品详情、用户 session、秒杀库存计数器。

Elasticsearch：商品搜索、日志分析（基于 Lucene 的搜索引擎，不是典型 NoSQL，但常与数据库配合）。

Cassandra：存储用户行为流水、点击流（写吞吐极高，可线性扩展）。

应用层通过业务规则决定数据写到哪里，并处理最终一致性问题（例如先写 MySQL，再异步同步到 Elasticsearch）。

第八部分：数据库进阶实战 —— 一个完整案例

我们以一个典型的“排名系统”为例，将上述知识串联起来。需求：一个游戏排行榜，记录每个玩家的得分，要求实时更新排名，并能高效查询前 100 名和玩家自己的排名。

8.1 方案一：关系数据库 + 索引优化
表结构：

CREATE TABLE scores (
    user_id INT PRIMARY KEY,
    score INT NOT NULL,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_score (score DESC)
);

查询前 100 名：

SELECT user_id, score FROM scores ORDER BY score DESC LIMIT 100;

查询某玩家排名：

SELECT COUNT(*) + 1 AS rank FROM scores WHERE score > (SELECT score FROM scores WHERE user_id = 123);

问题：COUNT 在大表下较慢；实时更新排名需要频繁写。适用于中等规模（千万以内）。

8.2 方案二：使用 Redis 有序集合

ZADD leaderboard 1500 playerA
ZADD leaderboard 2800 playerB
ZREVRANGE leaderboard 0 99 WITHSCORES    # 前100名
ZREVRANK leaderboard playerB             # 获取排名（0-based）

Redis 操作都是 O(log N)，百万玩家轻松应对，且天然支持实时更新。但 Redis 内存昂贵，数据量太大时成本高。可以冷热分离：热数据（活跃玩家）放 Redis，冷数据（历史玩家）放 MySQL。

8.3 方案三：基于数据库的分区+缓存
如果坚持使用 MySQL，可以按 score 范围分区，例如每 1000 分一个分区，这样查询前 100 名只需扫描高分区的前几页。同时使用 Redis 作为查询缓存，score 更新时使缓存失效。

8.4 完整代码实现（应用层逻辑）
用 Python 展示排行榜服务的实现，包含缓存和降级：

import redis
import pymysql
import json

class LeaderboardService:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.db = pymysql.connect(host='localhost', user='app', password='xxx', database='game')

    def update_score(self, user_id, new_score):
        # 先更新数据库
        with self.db.cursor() as cursor:
            cursor.execute("INSERT INTO scores (user_id, score) VALUES (%s, %s) ON DUPLICATE KEY UPDATE score = %s",
                           (user_id, new_score, new_score))
            self.db.commit()
        # 再更新 Redis
        self.redis_client.zadd('leaderboard', {user_id: new_score})
        # 删除该用户的缓存排名（懒加载）
        self.redis_client.delete(f'rank:{user_id}')

    def get_top_n(self, n=100):
        # 优先从 Redis 读取，不存在则从 MySQL 回源并写入缓存
        cached = self.redis_client.get(f'top_{n}')
        if cached:
            return json.loads(cached)
        with self.db.cursor() as cursor:
            cursor.execute("SELECT user_id, score FROM scores ORDER BY score DESC LIMIT %s", (n,))
            rows = cursor.fetchall()
            result = [{'user_id': r[0], 'score': r[1]} for r in rows]
            # 缓存 60 秒，避免频繁穿透
            self.redis_client.setex(f'top_{n}', 60, json.dumps(result))
            # 同时预热 Redis ZSET（如果 Redis 内存允许）
            for r in rows:
                self.redis_client.zadd('leaderboard', {r[0]: r[1]})
        return result

    def get_user_rank(self, user_id):
        # 先尝试从 Redis ZSET 获取
        rank = self.redis_client.zrevrank('leaderboard', user_id)
        if rank is not None:
            return rank + 1
        # 降级查数据库
        with self.db.cursor() as cursor:
            cursor.execute("SELECT COUNT(*) + 1 FROM scores WHERE score > (SELECT score FROM scores WHERE user_id=%s)", (user_id,))
            rank = cursor.fetchone()[0]
            # 回填到 Redis
            self.redis_client.zadd('leaderboard', {user_id: self._get_user_score_from_db(user_id)})
        return rank

请记住：数据库是系统的核心，任何疏忽都可能酿成严重事故。每次对数据库的变更（加索引、改表结构、调参数）都应该在测试环境验证，并有灰度上线和回滚方案。以敬畏之心对待数据，以科学之眼分析性能，你就能成为真正的数据库专家。
来源：
https://yvyus.cn/