综合练习

这一课你将能做到

1. 为「在线判题学习平台」设计一套 MongoDB 文档模型，包含用户、课时、提交记录、排行榜四个集合
2. 用索引优化三个高频查询，并用聚合管道输出一张「课时通过率 + 用户排名」双报表
3. 用 Change Streams 监听新提交，触发排行榜实时更新，形成闭环

为什么要学

前 11 课是零件，这一课是整车。学习平台的数据形状天然适合 MongoDB：提交记录是文档，排行榜要实时，用户进度要快速更新。把 11 课的知识点串成一个可运行的最小系统，才算真正学会。

先建立图像

数据模型（4 个集合）：

users          lessons            submissions            leaderboard
├─ _id         ├─ _id             ├─ _id                 ├─ _id (userId)
├─ nickname    ├─ slug            ├─ userId              ├─ nickname
├─ avatar      ├─ title           ├─ lessonSlug          ├─ totalScore
└─ createdAt   ├─ difficulty      ├─ status              ├─ passCount
               ├─ tags[]          ├─ score               ├─ lastSubmitAt
               └─ trackSlug       ├─ attempts            └─ updatedAt
                                  ├─ cases[]（内嵌）
                                  └─ submittedAt

数据流：
提交 → submissions (insert) → Change Stream → 累加 leaderboard (upsert + $inc)
                                            ↘ Kafka → 异步通知/风控
查询排行榜 → leaderboard 单集合（无 JOIN） → 按 totalScore 倒序取前 10

跟着做

步骤 1：创建集合与初始数据

use codeforge

// 课时字典
db.lessons.drop()
db.lessons.insertMany([
  { slug: "mdbrm-intro",    title: "MongoDB 概览",   difficulty: 1, trackSlug: "mongodb-roadmap-essentials" },
  { slug: "mdbrm-types",    title: "BSON 与类型",    difficulty: 2, trackSlug: "mongodb-roadmap-essentials" },
  { slug: "mdbrm-crud",     title: "CRUD 操作",      difficulty: 2, trackSlug: "mongodb-roadmap-essentials" },
  { slug: "mdbrm-query",    title: "查询运算符",     difficulty: 3, trackSlug: "mongodb-roadmap-essentials" },
  { slug: "mdbrm-agg",      title: "聚合管道",       difficulty: 3, trackSlug: "mongodb-roadmap-essentials" }
])

// 用户
db.users.drop()
db.users.insertMany([
  { _id: 1024, nickname: "小林", avatar: "a.png", createdAt: new Date("2026-09-01") },
  { _id: 2048, nickname: "阿明", avatar: "b.png", createdAt: new Date("2026-09-15") },
  { _id: 4096, nickname: "老王", avatar: "c.png", createdAt: new Date("2026-10-01") }
])

// 提交记录（带内嵌 cases）
db.submissions.drop()
db.submissions.insertMany([
  { userId: 1024, lessonSlug: "mdbrm-intro", status: "accepted", score: 100, attempts: 1,
    cases: [{name:"c1",passed:true,costMs:12},{name:"c2",passed:true,costMs:8}],
    submittedAt: new Date("2026-10-01T08:00:00Z") },
  { userId: 1024, lessonSlug: "mdbrm-types", status: "failed",   score: 40,  attempts: 2,
    cases: [{name:"c1",passed:true,costMs:15},{name:"c2",passed:false,costMs:300}],
    submittedAt: new Date("2026-10-02T09:00:00Z") },
  { userId: 1024, lessonSlug: "mdbrm-types", status: "accepted", score: 88,  attempts: 1,
    cases: [{name:"c1",passed:true,costMs:10},{name:"c2",passed:true,costMs:25}],
    submittedAt: new Date("2026-10-03T10:00:00Z") },
  { userId: 2048, lessonSlug: "mdbrm-intro", status: "accepted", score: 95,  attempts: 1,
    cases: [{name:"c1",passed:true,costMs:11},{name:"c2",passed:true,costMs:9}],
    submittedAt: new Date("2026-10-04T11:00:00Z") },
  { userId: 2048, lessonSlug: "mdbrm-crud",  status: "accepted", score: 78,  attempts: 2,
    cases: [{name:"c1",passed:true,costMs:20},{name:"c2",passed:false,costMs:180}],
    submittedAt: new Date("2026-10-05T14:00:00Z") },
  { userId: 4096, lessonSlug: "mdbrm-intro", status: "failed",   score: 30,  attempts: 3,
    cases: [{name:"c1",passed:false,costMs:200},{name:"c2",passed:false,costMs:400}],
    submittedAt: new Date("2026-10-06T16:00:00Z") }
])

// 排行榜（初始为空，由 Change Stream 或聚合初始化）
db.leaderboard.drop()

步骤 2：建立索引

按 ESR 法则：

// 查询 1：查某用户的提交列表（等值 + 排序）
db.submissions.createIndex({ userId: 1, submittedAt: -1 })

// 查询 2：查某课的通过情况（等值 + 等值 + 范围）
db.submissions.createIndex({ lessonSlug: 1, status: 1, score: -1 })

// 查询 3：排行榜单集合查询（点查 + 排序）
db.leaderboard.createIndex({ totalScore: -1, lastSubmitAt: -1 })

验证查询 1：

db.submissions.find({ userId: 1024 }).sort({ submittedAt: -1 }).explain("executionStats")
// stage: IXSCAN, docsExamined ≈ 3（该用户只有 3 条）

步骤 3：聚合报表 —— 课时通过率

db.submissions.aggregate([
  // 按课分组：总提交数、通过数、平均分
  { $group: {
      _id: "$lessonSlug",
      total: { $sum: 1 },
      passCount: { $sum: { $cond: [{ $eq: ["$status", "accepted"] }, 1, 0] } },
      avgScore: { $avg: "$score" }
  }},
  // 关联课时标题
  { $lookup: {
      from: "lessons",
      localField: "_id",
      foreignField: "slug",
      as: "lesson"
  }},
  { $unwind: "$lesson" },
  // 重塑输出
  { $project: {
      _id: 0,
      lessonSlug: "$_id",
      title: "$lesson.title",
      difficulty: "$lesson.difficulty",
      total: 1,
      passCount: 1,
      passRate: { $round: [{ $multiply: [{ $divide: ["$passCount", "$total"] }, 100] }, 1] },
      avgScore: { $round: ["$avgScore", 1] }
  }},
  { $sort: { passRate: -1 } }
])

输出示例：

{ lessonSlug: "mdbrm-intro", title: "MongoDB 概览", passRate: 66.7, avgScore: 75, total: 3, passCount: 2 }

步骤 4：聚合报表 —— 用户排名

db.submissions.aggregate([
  { $match: { status: "accepted" } },
  { $group: {
      _id: "$userId",
      totalScore: { $sum: "$score" },
      passCount: { $sum: 1 },
      lastSubmitAt: { $max: "$submittedAt" }
  }},
  { $sort: { totalScore: -1, lastSubmitAt: 1 } },   // 同分时先提交者在前
  { $lookup: { from: "users", localField: "_id", foreignField: "_id", as: "user" } },
  { $unwind: "$user" },
  { $project: {
      _id: 0, userId: "$_id", nickname: "$user.nickname",
      totalScore: 1, passCount: 1, lastSubmitAt: 1
  }},
  { $limit: 10 }
])

步骤 5：Change Stream 实时更新排行榜

// 消费者：监听 accepted 提交，实时累加 leaderboard
const watcher = db.submissions.watch([
  { $match: { operationType: "insert", "fullDocument.status": "accepted" } }
])

for await (const change of watcher) {
  const doc = change.fullDocument
  db.leaderboard.updateOne(
    { _id: doc.userId },
    {
      $inc: { totalScore: doc.score, passCount: 1 },
      $set: { lastSubmitAt: doc.submittedAt, updatedAt: new Date() },
      $setOnInsert: { nickname: "unknown" }   // 首次插入时占位
    },
    { upsert: true }
  )
  // 异步把 nickname 补全
  const user = db.users.findOne({ _id: doc.userId }, { nickname: 1 })
  if (user) db.leaderboard.updateOne({ _id: doc.userId }, { $set: { nickname: user.nickname } })
}

测试：插入一条新 accepted 提交，排行榜立刻有数据：

db.submissions.insertOne({
  userId: 4096, lessonSlug: "mdbrm-types", status: "accepted", score: 85, attempts: 1,
  cases: [{name:"c1",passed:true,costMs:15}], submittedAt: new Date()
})
// Change Stream 触发后：
db.leaderboard.find().sort({ totalScore: -1 })

步骤 6：验证与备份演练

// 验证数据一致性：排行榜分数 = 该用户所有 accepted 提交分数之和
db.submissions.aggregate([
  { $match: { status: "accepted", userId: 4096 } },
  { $group: { _id: null, sum: { $sum: "$score" } } }
])
// 对比 leaderboard 中 4096 的 totalScore

备份：

docker exec m1 mongodump --db codeforge --collection leaderboard --out /data/backup

讲清楚要点

1. 排行榜冗余设计：totalScore 是从 submissions 派生的「物化视图」，靠 Change Stream 实时维护。直接用聚合每次算也行，但用户量大了会慢——冗余是读多写少场景的常规手段。
2. 内嵌 vs 引用的边界：submissions.cases 内嵌（一起读）；submissions.lessonSlug 是引用（标题等字典信息变化少，没必要内嵌冗余）。
3. 聚合报表的「物化」策略：步骤 3 的课时通过率可以 $merge 到 lessonStats 集合，让后台页面直接查结果集合而不是每次跑聚合。报表类场景优先物化。
4. Change Stream 的幂等性：同一事件可能被消费多次（网络重试、resume 位置不精确），所以 $inc 而不是 $set——如果同一事件重复到达，加两次 85 分就是错的。生产环境要用「事件唯一 ID」去重（_id 就是天然幂等键）。
5. 数据一致性校验：步骤 6 的校验脚本应定期跑（cron），发现不一致时以 submissions 为准重算 leaderboard。冗余数据必须有 reconciliation 机制。

你可能会懵的地方

• 排行榜为什么不直接用聚合每次算？ 数据量小可以；用户 10 万、提交 1000 万时，每次首页刷新都跑聚合太重。用 Change Stream 预聚合到 leaderboard 集合，查询变成单集合点查。
• Change Stream 消费失败怎么办？ try/catch + 重试 + 告警；resume token 持久化；消费幂等（用 _id 去重）。消费端挂掉期间的事件会在恢复后重放（只要 oplog 没过期）。
• leaderboard 里 nickname 冗余了，用户信息改了怎么办？ 再开一个 Change Stream 监听 users 集合的 nickname 更新，同步刷新 leaderboard。或者接受最终一致性，在展示层 join 查 users。
• 能不能用 $merge 替代 Change Stream？ 可以：定时跑聚合 $merge 到 leaderboard。但延迟高（分钟级），适合日报类；实时排行榜必须 Change Stream。

错法对照

错法	问题	正解
排行榜每次实时聚合 submissions	数据量大后首页卡死	Change Stream 预聚合到 leaderboard
Change Stream 消费者用 $set 累加分数	重复事件导致分数错	$inc + _id 幂等去重
submissions 里冗余 lesson title	标题改一次要改 10 万条	引用 + $lookup 或应用层拼接
排行榜无 reconciliation	数据漂移永远发现不了	定期聚合校验并重算

卡点急救

1. Change Stream 触发但 leaderboard 没更新：检查 for await 循环是否阻塞在上一轮（上一个事件处理异常没捕获）；确认 upsert 没报 duplicate key（_id 冲突）。
2. 聚合报表 passRate 全是 0：$divide 的两个操作数类型不匹配（int/int 得 int），用 $multiply 转 double 或 $toDouble。
3. 备份恢复后数据对不上：dump 期间有写入，加 --oplog 重放到一致点；或先停写再 dump。

面试常问

• 设计一个实时排行榜系统，你会怎么设计？ 数据源用 submissions 集合，Change Stream 监听 accepted 事件，消费者用 $inc 幂等更新 leaderboard 集合（_id 即 userId），查询走 leaderboard 单集合索引排序。考虑 Change Stream 断线重连 + 定期对账。
• MongoDB 里怎么做「数据修正」（reconciliation）？ 写校验脚本对比源集合（submissions 聚合）与目标集合（leaderboard），不一致时以源为准重算；跑在低频 cron 里，带告警。
• 这个系统上量后瓶颈在哪？ Change Stream 单消费者是瓶颈；可改为 Kafka Connector → Kafka 分区消费 → 多 worker 更新 leaderboard（按 userId hash 分片）。ES 同步也可挂到 Kafka 下游。

自测三问

1. 排行榜为什么不用实时聚合而要用 Change Stream + 冗余集合？
2. Change Stream 消费端为什么要幂等？怎么实现幂等？
3. 课时通过率的聚合管道里，$cond 和 $divide 分别起了什么作用？

小结一句话

11 课的知识在这一课闭环：内嵌建模定结构、ESR 索引定性能、聚合管道定报表、Change Stream 定实时——MongoDB 的完整拼图，就是一套「写得快、查得快、流得快」的系统。

延伸

• MongoDB 官方示例应用（mflix）
• MongoDB University 实战课程
• 本 track 完结：回到 MongoDB 路线图 标记全部节点已掌握，或挑战 specialty-databases-essentials。