MongoDB 面试题

// 一个典型的 BSON 文档示例
{
  _id: ObjectId("507f1f77bcf86cd799439011"),  // 主键，12 字节：时间戳(4) + 机器(3) + PID(2) + 计数器(3)
  user_id: 10001,
  username: "zhangsan",
  profile: {                                    // 嵌套文档
    real_name: "张三",
    age: 28,
    address: {
      city: "Beijing",
      district: "Haidian"
    }
  },
  tags: ["vip", "tech", "finance"],             // 数组
  created_at: ISODate("2024-01-15T08:30:00Z"),
  balance: NumberDecimal("1234.56")             // Decimal128 精确数值
}

// 单文档插入
db.users.insertOne({
  username: "zhangsan",
  email: "[email protected]",
  age: 28,
  status: "active",
  created_at: new Date()
})

// 多文档插入（有序，性能更优）
db.users.insertMany([
  { username: "lisi", age: 25, status: "active" },
  { username: "wangwu", age: 30, status: "inactive" }
], { ordered: false })  // ordered=false 失败不中断后续

// insertOne 返回结构
// {
//   acknowledged: true,
//   insertedId: ObjectId("...")
// }

// 基础查询
db.users.find({ status: "active", age: { $gte: 18 } })

// 投影：仅返回指定字段
db.users.find(
  { status: "active" },
  { username: 1, email: 1, _id: 0 }
)

// 比较操作符
db.products.find({
  price: { $gt: 100, $lte: 500 },     // 100 < price <= 500
  stock: { $ne: 0 },                    // stock != 0
  category: { $in: ["electronics", "books"] }
})

// 逻辑操作符
db.users.find({
  $and: [
    { age: { $gte: 18 } },
    { $or: [
      { status: "active" },
      { vip_level: { $gt: 3 } }
    ]}
  ]
})

// 元素操作符
db.users.find({
  phone: { $exists: true, $type: "string" },
  tags: { $size: 3 },                   // 数组长度为 3
  profile: { $elemMatch: { key: "city", value: "Beijing" } }
})

// 数组操作符
db.posts.find({ tags: { $all: ["mongodb", "nosql"] } })  // 包含所有元素
db.posts.find({ "comments.author": "zhangsan" })          // 数组元素匹配

// $set 更新字段
db.users.updateOne(
  { _id: ObjectId("...") },
  {
    $set: { status: "inactive", last_login: new Date() },
    $inc: { login_count: 1 },           // 字段自增
    $push: { tags: "new_tag" },         // 数组追加
    $addToSet: { roles: "admin" },      // 数组去重追加
    $unset: { deprecated_field: "" }    // 删除字段
  }
)

// updateMany 批量更新
db.products.updateMany(
  { category: "books", stock: { $lt: 10 } },
  { $set: { low_stock_alert: true } }
)

// upsert：不存在则插入
db.users.updateOne(
  { username: "newuser" },
  { $set: { email: "[email protected]", created_at: new Date() } },
  { upsert: true }
)

// findAndModify：原子操作并返回文档
db.counters.findAndModify({
  query: { _id: "order_id" },
  update: { $inc: { seq: 1 } },
  new: true,                            // 返回更新后的文档
  upsert: true
})

// 删除单个文档
db.users.deleteOne({ _id: ObjectId("...") })

// 批量删除
db.logs.deleteMany({
  created_at: { $lt: new Date(Date.now() - 30 * 24 * 3600 * 1000) }  // 30 天前
})

// 危险操作：删除集合所有文档
db.users.deleteMany({})                // 等同于 TRUNCATE

// countDocuments vs estimatedDocumentCount
db.users.countDocuments({ status: "active" })     // 精确计数（扫描）
db.users.estimatedDocumentCount()                   // 估算（使用元数据，快）

// 排序、限制、跳过
db.users.find({ status: "active" })
  .sort({ created_at: -1 })              // 降序
  .limit(20)
  .skip(0)

// 基于游标的高效分页（避免 skip 大偏移量）
const cursor = db.users.find({ status: "active" })
  .sort({ _id: 1 })
  .limit(20)
const lastId = cursor[19]._id
const nextPage = db.users.find({
  status: "active",
  _id: { $gt: lastId }                   // 游标分页
}).limit(20)

// 单字段索引
db.users.createIndex({ email: 1 })       // 1: 升序, -1: 降序
db.users.createIndex({ created_at: -1 })

// 复合索引（遵循 ESR 原则：Equality, Sort, Range）
db.orders.createIndex(
  { user_id: 1, status: 1, created_at: -1 }
)
// 等值 user_id + 等值 status + 范围/排序 created_at 均可命中索引

// 唯一索引
db.users.createIndex({ email: 1 }, { unique: true })

// 部分索引：仅索引部分文档，节省空间
db.orders.createIndex(
  { created_at: -1 },
  { partialFilterExpression: { status: "active" } }
)

// 稀疏索引：跳过 null 字段
db.products.createIndex(
  { discontinued_at: 1 },
  { sparse: true }                       // 不索引 discontinued_at 为空的文档
)

// TTL 索引：自动过期删除
db.sessions.createIndex(
  { created_at: 1 },
  { expireAfterSeconds: 3600 }            // 1 小时后自动删除
)

// 文本索引：全文搜索
db.articles.createIndex(
  { title: "text", content: "text" },
  { weights: { title: 10, content: 1 } }
)
db.articles.find({ $text: { $search: "mongodb performance" } })

// 多键索引：数组字段自动创建
db.posts.createIndex({ tags: 1 })          // tags 为数组，自动 multikey
db.posts.find({ tags: "mongodb" })        // 命中索引

// 地理空间索引
db.locations.createIndex({ location: "2dsphere" })
db.locations.find({
  location: {
    $near: {
      $geometry: { type: "Point", coordinates: [116.404, 39.915] },
      $maxDistance: 5000
    }
  }
})

// 哈希索引（用于 Shard Key 哈希分片）
db.users.createIndex({ _id: "hashed" })

// 查询执行计划
db.orders.find({ user_id: 10001, status: "active" })
  .sort({ created_at: -1 })
  .explain("executionStats")             // 详细统计

// 关键字段解读

// 反例：全表扫描 COLLSCAN（性能差）
{
  "winningPlan": {
    "stage": "COLLSCAN",
    "filter": { "user_id": { "$eq": 10001 } }
  },
  "executionStats": {
    "totalDocsExamined": 1000000,
    "nReturned": 1,
    "executionTimeMillis": 850
  }
}

// 正例：索引扫描 IXSCAN（性能优）
{
  "winningPlan": {
    "stage": "FETCH",
    "inputStage": {
      "stage": "IXSCAN",
      "indexName": "user_id_1_status_1_created_at_-1"
    }
  },
  "executionStats": {
    "totalKeysExamined": 1,
    "totalDocsExamined": 1,
    "nReturned": 1,
    "executionTimeMillis": 2
  }
}

// 创建覆盖索引
db.orders.createIndex({ user_id: 1, status: 1, total: 1 })

// 查询完全命中索引（无需回表）
db.orders.find(
  { user_id: 10001, status: "active" },
  { user_id: 1, status: 1, total: 1, _id: 0 }    // 仅投影索引字段
)
// executionStats.stage = "IXSCAN"（无 FETCH 阶段），性能最佳

1. 客户端发起 insert/update
2. WiredTiger 修改内存中的 B+Tree（copy-on-write）
3. 写入预写日志（WAL） → journal 文件
4. journal 落盘（fsync）→ 确认写入成功
5. 异步将脏页刷盘（eviction + checkpoint）
6. Checkpoint 创建一致性快照（默认 60 秒一次）

# mongod.conf（WiredTiger 配置段）
storage:
  engine: wiredtiger
  wiredTiger:
    engineConfig:
      cacheSizeGB: 8                    # WiredTiger 缓存大小（默认 50% RAM - 1GB）
      journalCompressor: snappy         # WAL 压缩算法（none/snappy/zlib）
      directoryForIndexes: false        # 索引与数据分离目录
      maxCacheOverflowFileSizeGB: 0     # 溢出文件大小（0 表示禁用磁盘溢出）
    collectionConfig:
      blockCompressor: snappy           # 集合数据压缩（none/snappy/zlib）
      indexConfig:
        prefixCompression: true         # 索引前缀压缩
    indexConfig:
      prefixCompression: true

// WiredTiger 使用 MVCC 实现文档级并发控制
// 不同文档的写操作完全并行，无锁竞争
db.orders.updateOne({ _id: 1 }, { $set: { status: "paid" } })
db.orders.updateOne({ _id: 2 }, { $set: { status: "shipped" } })
// 上述两个操作可以并发执行，无需加锁

// 同一文档的并发写会串行化
// WiredTiger 在文档级别检测冲突，必要时短暂等待或重试

// WiredTiger 状态查看
db.serverStatus().wiredTiger
// 关注指标：
// - cache: "maximum bytes configured" / "bytes currently in the cache"
// - cache: "tracked dirty bytes in the cache"
// - cache: "pages read into cache" / "pages written from cache"
// - cache: "percentage overhead"
// - cache: "eviction server evicting pages"
// - "bytes belonging to page images in the cache"

// 初始化复制集
rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "mongo-1:27017", priority: 2 },     // 高优先级倾向
    { _id: 1, host: "mongo-2:27017", priority: 1 },
    { _id: 2, host: "mongo-3:27017", priority: 1 },
    { _id: 3, host: "mongo-arbiter:27017", arbiterOnly: true }
  ]
})

// 查看复制集状态
rs.status()
// 输出关键字段：
// - state: 1 (PRIMARY) / 2 (SECONDARY) / 7 (ARBITER)
// - members[].stateStr
// - members[].optimeDate: 最后应用的 oplog 时间
// - members[].health: 1 (健康) / 0 (宕机)
// - members[].electionDate: 上次选举时间

// 查看主从延迟
rs.printSecondaryReplicationInfo()
// 输出: "source: mongo-1, syncedTo: ..., lag: 0s"

oplog 是一组固定大小的 capped collection，存储在 local.oplog.rs，记录所有修改数据的操作。

关键特性：
- 幂等性：oplog 操作可重放（除 oplog 应用本身）
- 幂等写入：所有 oplog 操作都包含 _id
- 时序性：每条操作携带时间戳与唯一 ID
- 容量：默认 5% 磁盘（可通过 oplogSizeMB 调整）
- 复制单位：Secondary 持续拉取 Primary oplog 并应用

// 复制集配置参数
cfg = rs.conf()

cfg.settings = {
  electionTimeoutMillis: 10000,         // 选举超时（默认 10s）
  heartbeatIntervalMillis: 2000,        // 心跳间隔
  catchUpTimeoutMillis: 2000,           // 新 Primary 追平数据超时
  chainingAllowed: true,                // 允许 Secondary 链式复制（4.0+）
  // 多数派写入控制
  getLastErrorDefaults: { w: 1, wtimeout: 0 },
  // w: "majority" 表示多数节点确认后才返回
  getLastErrorModes: [
    { name: "multi-region", loc: { "$in": ["us", "eu", "ap"] } }
  ]
}

rs.reconfig(cfg)

// 写入确认级别
db.orders.insertOne(
  { user_id: 10001, total: 99.9 },
  { writeConcern: { w: "majority", wtimeout: 5000 } }  // 多数节点确认，5s 超时
)

// 写入关注级别
// w: 0     - 不等待确认（最快，可能丢失）
// w: 1     - Primary 确认（默认）
// w: 2     - 1 Primary + 1 Secondary 确认
// w: "majority" - 多数节点确认（推荐生产）
// w: <number> - 自定义数量

// 读关注（Read Concern）
db.orders.find({ user_id: 10001 })
  .readConcern("majority")              // 读取已确认的多数派数据
  .readPreference("primaryPreferred")   // 优先 Primary，失败读 Secondary

// 1. 启用分片
sh.enableSharding("mydb")

// 2. 选择 Shard Key 并创建索引
db.orders.createIndex({ user_id: 1 })

// 3. 对集合进行分片
sh.shardCollection(
  "mydb.orders",
  { user_id: 1 }                        // 范围分片
)
// 或
sh.shardCollection(
  "mydb.orders",
  { user_id: "hashed" }                 // 哈希分片
)

// 4. 复合分片
sh.shardCollection(
  "mydb.orders",
  { region: 1, user_id: 1 }             // 先 region，再 user_id
)

Chunk 生命周期：

1. 初始：每个 Shard 持有若干 Chunk
2. 拆分（Split）：当 Chunk 数据量超过 chunkSize（默认 128MB）触发拆分
3. 迁移（Migrate）：Balancer 检测到 Shard 间 Chunk 数量不均衡时
   触发迁移，源 Shard 发送 Chunk 数据至目标 Shard

Chunk 大小配置：
- 默认 128MB（适合大多数场景）
- 调整：db.settings.save({ _id: "chunksize", value: 64 })  // 单位 MB
- 大 Chunk：减少元数据，但拆分不灵活
- 小 Chunk：迁移频繁（容易影响性能）

// 查看 Balancer 状态
sh.getBalancerState()                    // true / false

// 关闭 Balancer（运维窗口）
sh.stopBalancer()

// 启用 Balancer
sh.startBalancer()

// 立即触发一次均衡（默认自动）
sh.startBalancer(timeout, interval)

// 监控 Chunk 分布
db.adminCommand({ listShards: 1 })
use config
db.chunks.aggregate([
  { $group: { _id: "$shard", count: { $sum: 1 } } }
])
// 输出：[{ _id: "shard-1", count: 50 }, { _id: "shard-2", count: 30 }]

// Targeted Query：包含 Shard Key 的查询，仅查询相关 Shard
db.orders.find({ user_id: 10001 })
// mongos 根据 user_id 路由到对应 Shard

// Broadcast Query：不包含 Shard Key 的查询，所有 Shard 都要查
db.orders.find({ status: "active" })
// mongos 向所有 Shard 广播，结果合并返回（性能差）

// 复合 Shard Key 部分命中
// Shard Key: { region: 1, user_id: 1 }
db.orders.find({ region: "us-east" })           // 部分命中，按 region 路由
db.orders.find({ user_id: 10001 })              // 不可路由，广播查询

// 典型聚合管道
db.orders.aggregate([
  // 1. 过滤：仅看 2024 年的订单
  {
    $match: {
      created_at: { $gte: ISODate("2024-01-01"), $lt: ISODate("2025-01-01") },
      status: "completed"
    }
  },
  // 2. 关联查询：获取用户信息
  {
    $lookup: {
      from: "users",
      localField: "user_id",
      foreignField: "_id",
      as: "user_info"
    }
  },
  // 3. 拆解数组：每个 user_info 元素作为独立文档
  { $unwind: "$user_info" },
  // 4. 投影：仅保留必要字段
  {
    $project: {
      _id: 0,
      order_id: "$_id",
      user_id: 1,
      total: 1,
      user_name: "$user_info.username",
      user_city: "$user_info.profile.address.city"
    }
  },
  // 5. 分组聚合：按城市统计 GMV
  {
    $group: {
      _id: "$user_city",
      total_gmv: { $sum: "$total" },
      order_count: { $sum: 1 },
      avg_order: { $avg: "$total" },
      max_order: { $max: "$total" },
      min_order: { $min: "$total" },
      unique_users: { $addToSet: "$user_id" }
    }
  },
  // 6. 计算字段：客单价
  {
    $addFields: {
      unique_user_count: { $size: "$unique_users" }
    }
  },
  // 7. 排序：GMV 降序
  { $sort: { total_gmv: -1 } },
  // 8. 限制：Top 10 城市
  { $limit: 10 }
])

// 单条件关联
db.orders.aggregate([
  {
    $lookup: {
      from: "users",                     // 关联集合
      localField: "user_id",             // 本集合字段
      foreignField: "_id",               // 关联集合字段
      as: "user"                         // 输出数组字段名
    }
  }
])

// 嵌套 pipeline 关联（4.0+）
db.orders.aggregate([
  {
    $lookup: {
      from: "users",
      let: { uid: "$user_id" },
      pipeline: [
        { $match: { $expr: { $eq: ["$_id", "$$uid"] } } },
        { $project: { username: 1, city: "$profile.address.city" } }
      ],
      as: "user"
    }
  }
])

// 多条件关联 + 相关子查询
db.orders.aggregate([
  {
    $lookup: {
      from: "products",
      let: { items: "$items" },
      pipeline: [
        { $match: { $expr: { $in: ["$_id", "$$items"] } } },
        { $project: { name: 1, price: 1 } }
      ],
      as: "product_details"
    }
  }
])

// 单次查询返回多个维度的统计
db.products.aggregate([
  {
    $facet: {
      // 价格区间分布
      price_ranges: [
        {
          $bucket: {
            groupBy: "$price",
            boundaries: [0, 100, 500, 1000, 5000, 10000],
            default: "10000+",
            output: {
              count: { $sum: 1 },
              avg_rating: { $avg: "$rating" }
            }
          }
        }
      ],
      // Top 10 评分
      top_rated: [
        { $sort: { rating: -1 } },
        { $limit: 10 },
        { $project: { name: 1, rating: 1, price: 1 } }
      ],
      // 类别统计
      by_category: [
        {
          $group: {
            _id: "$category",
            count: { $sum: 1 },
            avg_price: { $avg: "$price" }
          }
        },
        { $sort: { count: -1 } }
      ]
    }
  }
])

// allowDiskUse 启用磁盘临时文件
db.orders.aggregate(
  [...],
  { allowDiskUse: true }
)

// 聚合管道执行分析
db.orders.explain("executionStats").aggregate([...])
// 关注：
// - $cursor.queryPlanner.winningPlan
// - $cursor.executionStats.totalDocsExamined
// - $cursor.executionStats.executionTimeMillis

// 方式 1：回调 API（推荐，自动重试 TransientTransactionError）
const session = db.getMongo().startSession()
session.startTransaction({
  readConcern: { level: "snapshot" },
  writeConcern: { w: "majority" },
  readPreference: "primary"
})

try {
  const users = session.getDatabase("mydb").users
  const orders = session.getDatabase("mydb").orders
  
  // 业务逻辑：转账
  users.updateOne(
    { _id: 10001 },
    { $inc: { balance: -100 } },
    { session }
  )
  users.updateOne(
    { _id: 10002 },
    { $inc: { balance: 100 } },
    { session }
  )
  orders.insertOne(
    { from: 10001, to: 10002, amount: 100, created_at: new Date() },
    { session }
  )
  
  session.commitTransaction()
  console.log("事务提交成功")
} catch (error) {
  // 区分可重试错误与不可重试错误
  if (error.hasErrorLabel("TransientTransactionError")) {
    // 临时错误（锁冲突、复制集选举），可重试
    console.log("可重试的事务错误")
  } else if (error.hasErrorLabel("UnknownTransactionCommitResult")) {
    // 提交结果未知（网络问题），可重试提交
    console.log("提交结果不确定，需重试 commit")
  } else {
    // 业务错误，回滚
    console.log("业务错误，事务回滚")
  }
  session.abortTransaction()
} finally {
  session.endSession()
}

// 方式 2：withTransaction 自动重试
session.withTransaction(
  async () => {
    const users = session.getDatabase("mydb").users
    const orders = session.getDatabase("mydb").orders
    
    users.updateOne({ _id: 10001 }, { $inc: { balance: -100 } }, { session })
    users.updateOne({ _id: 10002 }, { $inc: { balance: 100 } }, { session })
    orders.insertOne(
      { from: 10001, to: 10002, amount: 100 },
      { session }
    )
  },
  {
    readConcern: { level: "snapshot" },
    writeConcern: { w: "majority" },
    readPreference: "primary"
  }
)

事务开销分析：

1. 时间戳管理：
   - 每个事务携带时间戳 t
   - WiredTiger MVCC 通过 t 确定可见性
   - 事务粒度越大，MVCC 维护成本越高

2. 锁竞争：
   - 多文档事务会持有 WiredTiger 内部锁
   - 锁冲突时等待时间随事务时长增加
   - 长事务会导致 WiredTiger cache 压力

3. 网络开销：
   - 事务开始 / 提交时需与所有相关节点通信
   - 跨分片事务需更多节点协调

4. 磁盘 I/O：
   - 事务回滚时产生 undo log
   - 大量小事务的 undo log 累积

// 查看事务状态
db.serverStatus().transactions
// 关键字段：
// - "transactionsCollectionWriterCount"
// - "transactionsCollectionReaderCount"
// - "totalAborted": 已回滚事务数
// - "totalCommitted": 已提交事务数
// - "totalStarted": 总启动事务数
// - "currentActive": 当前活跃事务数
// - "currentInactive": 当前空闲事务数

// 查看活跃事务
db.adminCommand({
  currentOp: 1,
  $or: [
    { "op": "command", "command.$_internalId": { $exists: true } },
    { "op": "command", "command.commitTransaction": { $exists: true } }
  ]
})

// MongoDB 单文档操作天然原子（无需事务）
// 适用：嵌入式结构内的字段更新
db.users.updateOne(
  { _id: 10001, balance: { $gte: 100 } },   // 乐观锁条件
  { $inc: { balance: -100 } }                // 条件性扣款
)
// 单文档原子操作性能更优，优先使用

// 多文档事务：跨集合 / 跨文档需要事务
// 如：账户 A 扣款 + 账户 B 加款 + 订单记录
// 必须使用事务保证三者一致

// Node.js 监听 orders 集合变更
const pipeline = [
  { $match: { "fullDocument.status": "active" } },  // 仅关注 active 订单
  { $project: { operationType: 1, fullDocument: 1, ns: 1, documentKey: 1 } }
]

const changeStream = db.collection("orders").watch(pipeline, {
  fullDocument: "updateLookup",            // update 时返回完整文档
  resumeAfter: { _data: "..." },            // 断点续传起点
  maxAwaitTimeMS: 1000,                    // 无变更时等待时间
  batchSize: 100
})

changeStream.on("change", (change) => {
  console.log("变更事件:", change.operationType)
  // insert / update / replace / delete / drop / rename / invalidate
  
  switch (change.operationType) {
    case "insert":
      handleNewOrder(change.fullDocument)
      break
    case "update":
      handleOrderUpdate(change.documentKey, change.fullDocument)
      break
    case "delete":
      handleOrderDelete(change.documentKey)
      break
  }
})

changeStream.on("error", (error) => {
  // 错误处理（可恢复 vs 不可恢复）
  if (error.code === 280) {  // ChangeStreamHistoryLost
    // Resume token 失效，需重新建立 watch
  }
})

// 关闭流
changeStream.close()

// insert 事件
{
  _id: { _data: "8265F8..." },          // resume token
  operationType: "insert",
  clusterTime: Timestamp(1700000000, 1),
  ns: { db: "mydb", coll: "orders" },
  documentKey: { _id: ObjectId("...") },
  fullDocument: { _id: ..., user_id: 10001, total: 99.9, status: "active" }
}

// update 事件（需 fullDocument: "updateLookup"）
{
  _id: { _data: "..." },
  operationType: "update",
  clusterTime: Timestamp(...),
  ns: { db: "mydb", coll: "orders" },
  documentKey: { _id: ObjectId("...") },
  updateDescription: {
    updatedFields: { status: "shipped" },
    removedFields: [],
    truncatedArrays: []
  },
  fullDocument: { _id: ..., status: "shipped", ... }  // 更新后完整文档
}

// delete 事件
{
  _id: { _data: "..." },
  operationType: "delete",
  clusterTime: Timestamp(...),
  ns: { db: "mydb", coll: "orders" },
  documentKey: { _id: ObjectId("...") }
}

// 1. 监听整个数据库
db.getSiblingDB("mydb").watch()

// 2. 监听整个集群
db.adminCommand({ aggregate: 1, pipeline: [...], cursor: {} })

// 3. 复杂过滤管道
const pipeline = [
  {
    $match: {
      $and: [
        { "ns.coll": { $in: ["orders", "payments"] } },
        {
          $or: [
            { operationType: "insert", "fullDocument.total": { $gt: 1000 } },
            { operationType: "update", "updateDescription.updatedFields.status": "refunded" },
            { operationType: "delete" }
          ]
        }
      ]
    }
  }
]

// 4. 断点续传（生产必备）
let resumeToken = null

const changeStream = db.orders.watch(pipeline, {
  resumeAfter: resumeToken               // 从上次中断点继续
})

changeStream.on("change", (change) => {
  resumeToken = change._id               // 保存 resume token
  // 处理变更
  processChange(change)
})

// 5. 配合 dead letter queue
changeStream.on("change", async (change) => {
  try {
    await processChange(change)
  } catch (error) {
    await deadLetterQueue.push({ change, error: error.message })
  }
})

# MongoDB Kafka Connector 配置示例
# 从 MongoDB 同步变更到 Kafka
name: mongodb-source-orders
config:
  connector.class: com.mongodb.kafka.connect.MongoSourceConnector
  connection.uri: mongodb://mongo-1:27017,mongo-2:27017,mongo-3:27017/?replicaSet=rs0
  database: mydb
  collection: orders
  # 复制集与 Change Stream 配置
  change.stream.full.document: updateLookup
  # 输出到 Kafka topic
  topic.prefix: mongodb.
  # Pipeline 过滤
  pipeline: |
    [
      { "$match": { "operationType": { "$in": ["insert", "update", "delete"] } } }
    ]
  # 起始位置
  startup.mode: resume

// Node.js GridFS Bucket API（推荐）
const { MongoClient, GridFSBucket } = require("mongodb")
const client = new MongoClient("mongodb://localhost:27017")

// 打开 GridFS Bucket
const db = client.db("mydb")
const bucket = new GridFSBucket(db, {
  bucketName: "fs",                      // 集合前缀（fs.files / fs.chunks）
  chunkSizeBytes: 261120                 // 255KB（默认）
})

// 1. 上传文件
const fs = require("fs")
const uploadStream = bucket.openUploadStream("video.mp4", {
  metadata: { category: "tutorial", uploaded_by: "zhangsan" }
})

fs.createReadStream("./local-video.mp4")
  .pipe(uploadStream)
  .on("error", (err) => console.error("上传失败:", err))
  .on("finish", () => {
    console.log("上传完成, fileId:", uploadStream.id)
  })

// 2. 下载文件
const downloadStream = bucket.openDownloadStreamByName("video.mp4")
downloadStream.pipe(fs.createWriteStream("./downloaded.mp4"))

// 3. 通过 _id 下载
const fileId = ObjectId("...")
bucket.openDownloadStream(fileId)
  .pipe(fs.createWriteStream("./by-id.mp4"))

// 4. 列出文件
const cursor = bucket.find({ "metadata.category": "tutorial" })
cursor.forEach(file => {
  console.log(`${file.filename} - ${file.length} bytes`)
})

// 5. 删除文件
bucket.delete(fileId)

// 6. 范围下载（部分内容）
bucket.openDownloadStreamByName("video.mp4", {
  start: 1024,                           // 起始字节
  end: 2048                              // 结束字节
})

// fs.files 集合 - 文件元数据
{
  _id: ObjectId("..."),
  filename: "presentation.pptx",
  length: 5242880,                        // 文件总大小
  chunkSize: 261120,                      // chunk 大小
  uploadDate: ISODate("2024-01-15T10:00:00Z"),
  md5: "5d41402abc4b2a76b9719d911017c592",
  contentType: "application/vnd.openxmlformats-officedocument.presentationml.presentation",
  metadata: {
    uploaded_by: "zhangsan",
    department: "engineering",
    tags: ["internal", "training"]
  },
  aliases: ["deck.pptx"]
}

// fs.chunks 集合 - 二进制块
{
  _id: ObjectId("..."),
  files_id: ObjectId("..."),             // 关联到 fs.files._id
  n: 0,                                  // chunk 序号（0 起始）
  data: BinData(0, "...")                 // 255KB 二进制数据
}

// 自定义 GridFS Bucket
const bucket = new GridFSBucket(db, {
  bucketName: "attachments",              // 自定义前缀
  chunkSizeBytes: 1024 * 1024,            // 1MB（适合大视频）
  writeConcern: { w: "majority" },        // 多数派写入
  readPreference: "primaryPreferred"      // 读偏好
})

// 优化索引
db.fs.chunks.createIndex({ files_id: 1, n: 1 })            // 默认索引
db.fs.files.createIndex({ "metadata.department": 1 })      // 业务查询索引
db.fs.files.createIndex({ uploadDate: -1 })                // 按时间清理

// 定期清理过期文件
db.fs.files.aggregate([
  { $match: { uploadDate: { $lt: new Date(Date.now() - 90 * 86400000) } } },
  { $group: { _id: null, file_ids: { $push: "$_id" } } }
]).forEach(result => {
  result.file_ids.forEach(id => {
    bucket.delete(id)
  })
})

# 全库备份
mongodump --host "rs0/mongo-1:27017,mongo-2:27017" \
          --port 27017 \
          --username backup_user \
          --password "${MONGO_PASSWORD}" \
          --authenticationDatabase admin \
          --gzip \                            # 启用压缩
          --archive=mongo-backup-20240115.gz  # 单文件归档

# 单数据库备份
mongodump --host rs0/mongo-1:27017 \
          --db mydb \
          --gzip \
          --archive=mydb-backup-20240115.gz

# 单集合备份
mongodump --host rs0/mongo-1:27017 \
          --db mydb \
          --collection orders \
          --gzip \
          --archive=orders-backup-20240115.gz

# 指定字段查询备份（部分数据）
mongodump --host rs0/mongo-1:27017 \
          --db mydb \
          --collection orders \
          --query '{"created_at": {"$gte": {"$date": "2024-01-01T00:00:00Z"}}}' \
          --gzip \
          --archive=orders-2024.gz

# 并行备份（加速）
mongodump --host rs0/mongo-1:27017 \
          --numParallelCollections 4 \      # 4 集合并行
          --gzip \
          --archive=mongo-backup-20240115.gz

# 恢复全库
mongorestore --host rs0/mongo-1:27017 \
             --username restore_user \
             --password "${MONGO_PASSWORD}" \
             --authenticationDatabase admin \
             --gzip \
             --archive=mongo-backup-20240115.gz \
             --drop                            # 恢复前删除已存在集合

# 恢复到不同数据库
mongorestore --host rs0/mongo-1:27017 \
             --nsInclude "mydb.*" \
             --nsFrom "mydb.*" --nsTo "mydb_restored.*" \
             --gzip \
             --archive=mongo-backup-20240115.gz

# 单集合恢复
mongorestore --host rs0/mongo-1:27017 \
             --db mydb \
             --collection orders \
             --gzip \
             --archive=orders-backup-20240115.gz

# 并行恢复
mongorestore --host rs0/mongo-1:27017 \
             --numParallelCollections 4 \
             --numInsertionWorkersPerCollection 8 \
             --gzip \
             --archive=mongo-backup-20240115.gz

# 索引重建（恢复时禁用索引构建，最后批量构建）
mongorestore --host rs0/mongo-1:27017 \
             --noIndexRestore \
             --gzip \
             --archive=mongo-backup-20240115.gz
# 恢复后单独构建索引
mongosh "mongodb://rs0/mongo-1:27017" --eval "
  db.orders.createIndex({ user_id: 1, created_at: -1 })
"

# 1. 开启 oplog 监听（持续备份 oplog 到文件）
mongooplog --host rs0/mongo-1:27017 \
           --from rs0/mongo-1:27017 \
           --seconds 86400 \                # 监听 24 小时
           > /backup/oplog-20240115.bson

# 2. 备份后应用 oplog（Point-In-Time Recovery）
# 假设 14:00 备份后，14:30 发生误操作，需要恢复到 14:25 状态

# 步骤 1：恢复 14:00 全量备份
mongorestore --host rs0/mongo-1:27017 \
             --drop \
             --archive=mongo-backup-1400.gz

# 步骤 2：应用 oplog 到 14:25
mongorestore --host rs0/mongo-1:27017 \
             --oplogReplay \
             --oplogLimit "1705305900:1"    # oplog 时间戳（14:25 对应位置）

// 1. 查找目标时间点的 oplog 位置
// 查询 oplog，找到 14:25:00 之前的最后一个操作
use local
db.oplog.rs.find({
  ts: { $lte: Timestamp(1705305900, 1) }    // 14:25:00 的 Timestamp
}).sort({ ts: -1 }).limit(1)

// 2. 记录该 ts，作为 --oplogLimit 参数
// 3. 执行恢复（如上）

# 分片集群的备份需要协调每个 Shard + Config Server

# 1. 停止 Balancer（避免备份期间 Chunk 迁移）
mongosh "mongodb://mongos-1:27017" --eval "sh.stopBalancer()"

# 2. 备份 Config Server
mongodump --host config-1:27017 \
          --db config \
          --gzip \
          --archive=config-backup.gz

# 3. 备份每个 Shard（可并行）
mongodump --host shard-1-primary:27017 \
          --gzip \
          --archive=shard1-backup.gz &
mongodump --host shard-2-primary:27017 \
          --gzip \
          --archive=shard2-backup.gz &
mongodump --host shard-3-primary:27017 \
          --gzip \
          --archive=shard3-backup.gz &
wait

# 4. 恢复后重启 Balancer
mongosh "mongodb://mongos-1:27017" --eval "sh.startBalancer()"

# Kubernetes CronJob 备份示例
apiVersion: batch/v1
kind: CronJob
metadata:
  name: mongodb-backup
spec:
  schedule: "0 2 * * *"                  # 每天凌晨 2 点
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: backup
            image: mongo:7.0
            command:
            - /bin/sh
            - -c
            - |
              mongodump --host mongo-primary:27017 \
                        --username $BACKUP_USER \
                        --password $BACKUP_PASSWORD \
                        --authenticationDatabase admin \
                        --gzip \
                        --archive=/backup/mongo-$(date +%Y%m%d).gz
              
              # 上传到 S3
              aws s3 cp /backup/mongo-$(date +%Y%m%d).gz \
                        s3://mongo-backups/daily/
              
              # 清理 30 天前本地备份
              find /backup -name "*.gz" -mtime +30 -delete
            env:
            - name: BACKUP_USER
              valueFrom:
                secretKeyRef:
                  name: mongo-credentials
                  key: username
            - name: BACKUP_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: mongo-credentials
                  key: password
          restartPolicy: OnFailure

// 启用 Profiler（记录慢查询）
db.setProfilingLevel(2)                   // 记录所有操作
db.setProfilingLevel(1, { slowms: 100 })  // 记录 > 100ms 的操作（推荐）

// 查看 Profiler 状态
db.getProfilingStatus()

// 分析慢查询
db.system.profile.find({
  millis: { $gt: 100 }                    // 慢查询阈值
}).sort({ ts: -1 }).limit(20).forEach(doc => {
  print(JSON.stringify({
    ts: doc.ts,
    op: doc.op,
    ns: doc.ns,
    millis: doc.millis,
    query: doc.command?.filter || doc.command?.query,
    planSummary: doc.planSummary,
    numYields: doc.numYield,
    docsExamined: doc.docsExamined,
    keysExamined: doc.keysExamined,
    nreturned: doc.nreturned,
    responseLength: doc.responseLength
  }, null, 2))
})

// 慢查询关键指标
// docsExamined / nreturned  → 索引效率（理想 = 1）
// keysExamined / nreturned  → 索引选择性
// numYields                → 锁等待次数
// millis                   → 总耗时

// 关闭 Profiler
db.setProfilingLevel(0)

// 反例 1：全表扫描（COLLSCAN）
db.orders.find({ status: "active", created_at: { $gte: ISODate("2024-01-01") } })
// 优化：创建复合索引
db.orders.createIndex({ status: 1, created_at: -1 })
// ESR 原则：等值 (status) → 范围/排序 (created_at)

// 反例 2：未使用覆盖索引
db.users.find({ email: "[email protected]" }, { username: 1, email: 1, _id: 0 })
// 优化：覆盖索引
db.users.createIndex({ email: 1, username: 1 })
// 查询时只读取索引，无 FETCH 阶段

// 反例 3：正则表达式前缀模糊查询
db.users.find({ name: /zhang/i })         // 无法使用索引
db.users.find({ name: /^zhang/ })        // 前缀匹配，可使用索引
// 前缀 /^value/ 命中索引，包含 /value/ 不命中

// 反例 4：$ne / $nin 操作符（索引效率低）
db.orders.find({ status: { $ne: "deleted" } })  // 索引扫描大部分
// 优化：使用 $in 替代
db.orders.find({ status: { $in: ["active", "pending", "shipped"] } })

// 反例 5：$exists 检查
db.users.find({ phone: { $exists: true } })  // 索引效率低
// 优化：使用 $type 替代
db.users.find({ phone: { $type: "string" } })

// 反例：宽表 + 大量字段
{
  _id: ObjectId("..."),
  order_id: "O20240115-001",
  user_basic: { name: "zhangsan", age: 28 },  // 用户基本信息
  user_profile: { ... },                        // 扩展信息
  items: [...],                                 // 订单商品
  shipping: { ... },                            // 物流信息
  payment: { ... },                             // 支付信息
  ...
}
// 文档过大（接近 16MB 上限），更新频繁导致局部更新失效

// 优化：反范式化拆分（按读写模式）
// 集合 1：orders（订单核心信息）
{
  _id: ObjectId("..."),
  order_id: "O20240115-001",
  user_id: 10001,
  status: "active",
  total: 99.9,
  created_at: ISODate("2024-01-15")
}

// 集合 2：order_items（订单商品）
{
  _id: ObjectId("..."),
  order_id: "O20240115-001",
  product_id: 20001,
  qty: 2,
  price: 49.95
}

// 集合 3：order_shipping（物流信息）
{
  _id: ObjectId("..."),
  order_id: "O20240115-001",
  carrier: "SF",
  tracking_no: "SF1234567890"
}

# mongod.conf 性能调优
storage:
  wiredTiger:
    engineConfig:
      cacheSizeGB: 12                    # 缓存 = (RAM - OS reserve) × 50%
      journalCompressor: snappy
    
    collectionConfig:
      blockCompressor: snappy            # 集合压缩
    
    indexConfig:
      prefixCompression: true            # 索引前缀压缩

# 系统层面
net:
  maxIncomingConnections: 2000          # 最大连接数

# 操作 Profiling
operationProfiling:
  mode: slowOp
  slowOpThresholdMs: 100                # 慢查询阈值

# 资源限制
processManagement:
  fork: true
  timeZoneInfo: /usr/share/zoneinfo

// 1. 实时 QPS / 延迟
db.serverStatus().opcounters
// - insert / query / update / delete / command / getmore

// 2. 连接数
db.serverStatus().connections
// - current / available / totalCreated / active

// 3. 复制状态
db.serverStatus().repl
// - replSetName / ismaster / secondary / primary

// 4. WiredTiger 缓存
db.serverStatus().wiredTiger.cache
// - "bytes currently in the cache" / "maximum bytes configured"
// - "tracked dirty bytes in the cache"
// - "pages read into cache" / "pages written from cache"
// - "percentage overhead"

// 5. 内存使用
db.serverStatus().mem
// - resident / virtual / mapped

// 6. 锁统计
db.serverStatus().locks
// - globalLock.totalTime / currentQueue

// serverStatus：综合运行状态
db.serverStatus()
// 关键字段：
// - host / process / pid / uptime / version
// - storageEngine: { name: "wiredTiger" }
// - connections: { current, available, totalCreated, active }
// - mem: { resident, virtual, mapped }
// - opcounters: { insert, query, update, delete, command, getmore }
// - opLatencies: { reads, writes, commands }
// - repl: { replSetName, ismaster, secondary, primary }
// - wiredTiger.cache: { ... }
// - locks: { ... }

// dbStats：集合 / 数据库统计
db.stats()
// 关键字段：
// - collections / objects / avgObjSize / dataSize / storageSize
// - indexes / indexSize
// - fsUsedSize / fsTotalSize

// collStats：单集合详细统计
db.orders.stats(1024)  // 1024 字节缩放
// 关键字段：
// - size / count / avgObjSize / maxSize
// - storageSize / totalIndexSize
// - wiredTiger: { ... } 详细存储信息

// currentOp：当前操作
db.currentOp({
  $or: [
    { "op": { $in: ["query", "update", "insert", "remove"] } },
    { "command.findAndModify": { $exists: true } }
  ],
  "secs_running": { $gt: 5 }             // 超过 5 秒
})

// 终止慢查询
db.killOp(opid)

// top：按集合统计访问量
db.adminCommand({ top: 1 })

// hostInfo：服务器信息
db.hostInfo()
// - system / os / extra

// 复制集状态
rs.status()
// 关键字段：
// - set: 集群名
// - myState: 1 (PRIMARY) / 2 (SECONDARY)
// - members[]:
//   - stateStr
//   - health
//   - optimeDate: 最后应用 oplog 时间
//   - electionDate
//   - syncSourceHost

// 主从延迟
rs.printSecondaryReplicationInfo()
// 输出：source: mongo-2, syncedTo: ..., lag: 2s

// oplog window：oplog 可回放时长
db.adminCommand({ replSetGetStatus: 1 }).members.map(m => ({
  name: m.name,
  state: m.stateStr,
  lag_seconds: (Date.now() - m.optimeDate?.getTime()) / 1000
}))

# mongodb_exporter 部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mongodb-exporter
spec:
  replicas: 1
  selector:
    matchLabels:
      app: mongodb-exporter
  template:
    metadata:
      labels:
        app: mongodb-exporter
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9216"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: exporter
        image: percona/mongodb_exporter:0.40
        args:
        - "--mongodb.uri=mongodb://mongo-exporter:${PASSWORD}@mongo-primary:27017/admin"
        - "--collect-all"
        - "--compatible-mode"
        ports:
        - containerPort: 9216
        env:
        - name: PASSWORD
          valueFrom:
            secretKeyRef:
              name: mongodb-credentials
              key: password

# Prometheus 告警规则示例
groups:
- name: mongodb_alerts
  rules:
  # 复制集 Primary 不可用
  - alert: MongoDBPrimaryDown
    expr: mongodb_rs_members_state == 1
    for: 30s
    labels:
      severity: critical
    annotations:
      summary: "MongoDB Primary 不可用"

  # 复制延迟
  - alert: MongoDBReplicationLag
    expr: mongodb_rs_members_optimeDate - mongodb_rs_members_primary_optimeDate > 30
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "MongoDB 复制延迟 > 30s"

  # WiredTiger 缓存使用率高
  - alert: MongoDBHighCacheUsage
    expr: mongodb_ss_wt_cache_bytes_currently_in_cache / mongodb_ss_wt_cache_maximum_bytes_configured > 0.9
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "WiredTiger 缓存使用率 > 90%"

  # 脏页率过高
  - alert: MongoDBHighDirtyPages
    expr: mongodb_ss_wt_cache_tracked_dirty_bytes / mongodb_ss_wt_cache_maximum_bytes_configured > 0.2
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "WiredTiger 脏页率 > 20%"

  # 连接数高
  - alert: MongoDBHighConnections
    expr: mongodb_ss_connections{conn_type="current"} > 1600
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "MongoDB 连接数 > 1600"

  # 备份缺失
  - alert: MongoDBNoRecentBackup
    expr: time() - mongodb_backup_last_success_time > 86400
    for: 1h
    labels:
      severity: critical
    annotations:
      summary: "MongoDB 24 小时内无成功备份"

# mtools：MongoDB 日志分析工具集
pip install mtools

# 常用 mloginfo / mlogfilter / mplotqueries
mloginfo mongo.log                          # 整体统计
mlogfilter mongo.log --slow 100             # 提取 > 100ms 慢查询
mlogfilter mongo.log --namespace mydb.orders # 提取特定集合
mplotqueries mongo.log --type scatter       # 慢查询可视化

MongoDB 性能基线计算公式：

内存需求：
  cacheSizeGB = (热数据 + 索引) × 1.3（30% 余量）
  推荐：cacheSizeGB ≤ RAM × 50% - 1GB
  系统内存建议：cacheSizeGB × 2.5

磁盘需求：
  数据大小：原始 JSON × 1.5（BSON 开销） × 1.3（压缩后） × 1.2（索引开销）
  加上 oplog（默认 5% 磁盘）
  加上 WiredTiger checkpoint 临时空间（≈ 数据量 2%）

CPU 需求：
  OLTP：每 1000 QPS 需约 2 vCPU
  复杂聚合：每 100 QPS 需约 4 vCPU

网络需求：
  复制流量：oplog 写入速率 × 副本数
  客户端流量：峰值 QPS × 平均响应体大小

-- MySQL：用户表（关系型）
CREATE TABLE users (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  username VARCHAR(50) UNIQUE NOT NULL,
  email VARCHAR(100) NOT NULL,
  age INT,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  INDEX idx_email (email),
  INDEX idx_created (created_at)
);

CREATE TABLE user_profiles (
  user_id BIGINT PRIMARY KEY,
  real_name VARCHAR(50),
  address JSON,                          -- 5.7+ 支持 JSON 类型
  FOREIGN KEY (user_id) REFERENCES users(id)
);

// MongoDB：用户文档（文档型）
db.users.insertOne({
  username: "zhangsan",
  email: "[email protected]",
  age: 28,
  created_at: new Date(),
  profile: {                              // 嵌入文档
    real_name: "张三",
    address: {
      city: "Beijing",
      district: "Haidian"
    }
  },
  tags: ["vip", "tech"]                   // 数组
})

// 索引
db.users.createIndex({ email: 1 }, { unique: true })
db.users.createIndex({ created_at: -1 })

-- MySQL：转账事务（标准 ACID）
START TRANSACTION;
UPDATE accounts SET balance = balance - 100 WHERE user_id = 10001;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 10002;
INSERT INTO transfers (from_user, to_user, amount) VALUES (10001, 10002, 100);
COMMIT;

// MongoDB：转账事务（4.0+）
const session = db.getMongo().startSession()
session.startTransaction()
try {
  const users = session.getDatabase("mydb").users
  const transfers = session.getDatabase("mydb").transfers
  
  users.updateOne({ _id: 10001 }, { $inc: { balance: -100 } }, { session })
  users.updateOne({ _id: 10002 }, { $inc: { balance: 100 } }, { session })
  transfers.insertOne({ from: 10001, to: 10002, amount: 100 }, { session })
  
  session.commitTransaction()
} catch (e) {
  session.abortTransaction()
} finally {
  session.endSession()
}

-- MySQL：JOIN 查询
SELECT u.id, u.username, COUNT(o.id) AS order_count
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
WHERE u.created_at >= '2024-01-01'
GROUP BY u.id, u.username
HAVING order_count > 0
ORDER BY order_count DESC
LIMIT 10;

// MongoDB：$lookup 实现（性能通常弱于 MySQL JOIN）
db.users.aggregate([
  { $match: { created_at: { $gte: ISODate("2024-01-01") } } },
  {
    $lookup: {
      from: "orders",
      localField: "_id",
      foreignField: "user_id",
      as: "orders"
    }
  },
  {
    $project: {
      username: 1,
      order_count: { $size: "$orders" }
    }
  },
  { $match: { order_count: { $gt: 0 } } },
  { $sort: { order_count: -1 } },
  { $limit: 10 }
])

是否需要强事务 + 复杂 SQL + 报表分析？
  ├── 是 → 优先 MySQL（或其他 RDBMS）
  └── 否 →
        数据是否结构化固定？
          ├── 是 → MySQL
          └── 否（Schema 多变）→
                写入量是否 > 1万 QPS？
                  ├── 是 → MongoDB（Sharding 水平扩展）
                  └── 否 →
                        是否需要灵活字段 / 嵌套结构？
                          ├── 是 → MongoDB
                          └── 否 → MySQL

生产环境常见组合：

MySQL  →  核心账务、交易、订单状态（强一致）
MongoDB →  商品详情、用户画像、日志、IoT 数据（半结构化）
Redis   →  缓存层、Session、计数器
ClickHouse →  OLAP 分析、日志分析
Elasticsearch →  全文搜索、日志检索

各司其职，通过数据同步工具（Debezium / DataX / Canal）打通

某电商平台订单服务偶发查询延迟飙升至 5s+，业务高峰期出现报警。

监控数据：
- WiredTiger 缓存使用率持续 > 95%
- slow query 数量激增
- 多个查询 explain 显示 COLLSCAN

// 1. 确认缓存状态
db.serverStatus().wiredTiger.cache
// {
//   "bytes currently in the cache": 13600000000,    // 12.6GB
//   "maximum bytes configured": 13800000000,        // 12.9GB
//   "tracked dirty bytes in the cache": 2700000000,
//   "pages read into cache": 15230000,                // 持续从磁盘读取
//   "percentage overhead": 7
// }

// 2. 定位慢查询
db.system.profile.find({ millis: { $gt: 1000 } }).sort({ ts: -1 }).limit(10)
// 发现：db.orders.find({ status: "active" }) 频繁出现

// 3. 检查索引
db.orders.getIndexes()
// 已有：{ user_id: 1, created_at: -1 }
// 缺少：{ status: 1 }

// 4. 验证：执行计划
db.orders.find({ status: "active" }).explain("executionStats")
// winningPlan.stage = "COLLSCAN"
// totalDocsExamined = 5000000, nReturned = 80000

1. 业务变更新增"按订单状态查询"功能，但未创建 status 索引
2. status 字段低基数（仅 5 个值），命中索引时选择度低
3. 但因 orders 集合热数据 > WiredTiger cacheSize（12.9GB）
4. 缓存淘汰导致索引频繁失效
5. 查询退化为 COLLSCAN，性能急剧下降

// 1. 创建复合索引（ESR 原则：等值 status + 排序 created_at）
db.orders.createIndex(
  { status: 1, created_at: -1 },
  { background: true }                   // 后台构建，不阻塞
)

// 2. 优化查询条件（添加时间范围）
db.orders.find({
  status: "active",
  created_at: { $gte: ISODate("2024-01-01") }
}).sort({ created_at: -1 }).limit(20)

// 3. 扩容 WiredTiger 缓存
storage:
  wiredTiger:
    engineConfig:
      cacheSizeGB: 16                    // 从 12GB 扩容至 16GB

某分片集群 Shard-1 数据量达 3TB（其他 Shard 仅 1TB），磁盘使用率告警。
查询 Shard-1 上订单的延迟显著高于其他 Shard。

// 1. 查看 Chunk 分布
use config
db.chunks.aggregate([
  { $group: { _id: { shard: "$shard", ns: "$ns" }, count: { $sum: 1 }, total: { $sum: "$size" } } },
  { $match: { "total": { $gt: 100 } } },
  { $sort: { "total": -1 } }
])
// 发现：shard-1 上某 Chunk 异常大（jumbo flag: true）

// 2. 查看 Jumbo Chunk
db.chunks.find({ jumbo: true }).count()
// 100+ jumbo chunks

// 3. 查看 Shard Key
db.collections.find({ _id: "mydb.orders" })
// key: { region: 1, user_id: 1 }

1. 业务初期数据按 region 范围分片
2. 单一 region（"us-east"）数据量激增（占总量 70%+）
3. region 内 user_id 范围分片
4. 但 us-east 内 user_id 集中在某段（业务用户群特征）
5. Shard Key 选择不当，导致数据倾斜
6. 出现 Jumbo Chunk（无法继续拆分）

// 步骤 1：紧急手动拆分 Jumbo Chunk
use config

// 查找 us-east 大 Chunk
db.chunks.find({
  ns: "mydb.orders",
  "min.region": "us-east",
  jumbo: true
}).forEach(chunk => {
  // 强制拆分（注意：在低峰期执行）
  sh.splitAt("mydb.orders", chunk.max)
})

// 步骤 2：迁移拆分后的 Chunk
sh.startBalancer()

// 步骤 3：长期方案 - 调整 Shard Key
// 3.1 评估业务查询模式
// 3.2 业务侧实现应用层 Shard Key 写入新集群
// 3.3 双写双读切换
// 3.4 老集群下线

// 长期方案：使用 hashed 分片或更细粒度 Shard Key
// 例如：{ region: 1, user_id: "hashed" }  // region 范围 + user_id 哈希

某团队使用 Change Stream 同步订单数据到 Kafka，出现"resume token 失效"错误。
Change Stream 自动断开，应用程序告警。

// 1. 查看 oplog 状态
rs.printReplicationInfo()
// oplog 起始与结束时间窗口

// 2. 计算 oplog window
use local
db.oplog.rs.stats()
// {
//   "maxSize": 51200000000,    // 50GB
//   "size": 51200000000,
//   "count": 15000000,
//   ...
// }

// 3. 查看 oplog window
db.adminCommand({ replSetGetStatus: 1 }).members.forEach(m => {
  print(m.name, m.optimeDate)
})
// 输出：mongo-1 2024-01-15 10:00:00  mongo-2 2024-01-15 09:55:00
// oplog window = 5min（过短）

// 4. 查看 oplog 写入速率
db.oplog.rs.aggregate([
  { $group: { _id: null, avg_size: { $avg: "$size" } } }
])

1. oplogSizeMB 默认是 5% 磁盘空间
2. 业务高写入速率（10万 ops/s）导致 oplog 快速填满
3. Secondary 节点追赶不及时，oplog window 缩短
4. Change Stream 消费速率跟不上 oplog 产生速率
5. resume token 对应的 oplog 已被覆盖 → Change Stream 失效

// 1. 紧急：调整 oplog 大小（需重启）
// mongod.conf
replication:
  oplogSizeMB: 102400                // 100GB（从 50GB 扩容）

// 重启 mongod（先 Secondary，再 Primary）
// 注意：oplogSizeMB 调整后需要重启生效

// 2. 优化 Change Stream 消费
// 2.1 增加消费者并发
changeStream.on("change", async (change) => {
  // 异步处理
  await processChangeAsync(change)
})

// 2.2 批量处理
let buffer = []
changeStream.on("change", (change) => {
  buffer.push(change)
  if (buffer.length >= 100) {
    processBatch(buffer)
    buffer = []
  }
})

// 3. 监控告警
// oplog window < 1h 触发告警
// Change Stream 断开立即告警

业务反馈部分查询返回空数据，但实际数据存在。
多次重试后查询成功，问题偶发。

// 1. 检查 Balancer 状态
sh.status()
// 显示最近一次 Chunk 迁移

// 2. 查看 mongos 日志
// 出现 "could not find chunk" 错误

// 3. 查看 mongos 缓存
db.adminCommand({ getParameter: 1, userCacheInvalidationIntervalSecs: 1 })
// mongos 缓存更新默认 1s，业务端查询 < 1s 命中陈旧路由

1. Balancer 正在迁移 Chunk
2. mongos 缓存的路由信息未及时更新
3. 查询命中陈旧路由，定位到错误 Shard
4. 查询返回空或文档不存在

// 1. 临时 flush mongos 缓存（不重启）
db.adminCommand({ flushRouterConfig: 1 })

// 2. 优化 Balancer 时间窗口（业务低峰期迁移）
use config
db.settings.updateOne(
  { _id: "balancer" },
  { $set: { activeWindow: { start: "02:00", stop: "06:00" } } }
)

// 3. 应用层重试机制
async function findWithRetry(collection, filter, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await collection.find(filter).toArray()
    } catch (err) {
      if (err.codeName === "StaleConfig" && i < maxRetries - 1) {
        await sleep(100)                  // 短暂等待
        continue
      }
      throw err
    }
  }
}

// 4. 监控告警
// - StaleConfig 错误率
// - 路由缓存命中率
// - Balancer 迁移频率

某 3 节点复制集在网络抖动时出现多次 Primary 切换，每次切换导致短暂写入失败。

// 1. 查看复制集状态历史
db.adminCommand({ replSetGetStatus: 1 }).members.forEach(m => {
  print(m.name, "electionDate:", m.electionDate, "state:", m.stateStr)
})

// 2. 查看 oplog 窗口
rs.printReplicationInfo()

// 3. 查看节点日志
// 出现 "Stepped down" / "Heartbeat failure" / "Election"

1. 节点分布在 3 个 AZ，AZ 间网络抖动
2. 节点间心跳延迟 > electionTimeoutMillis（10s）
3. 触发不必要的选举
4. 新 Primary 提升后，旧 Primary 网络恢复又重新加入
5. 多次切换导致服务抖动

// 1. 调整选举参数
cfg = rs.conf()
cfg.settings = {
  electionTimeoutMillis: 20000,         // 增加到 20s（容忍网络抖动）
  heartbeatIntervalMillis: 2000,
  catchUpTimeoutMillis: 30000
}
rs.reconfig(cfg)

// 2. 优化节点分布
// - Primary 与多数派节点在同 AZ
// - 少数派节点用于灾备（Priority 0）

cfg.members = [
  { _id: 0, host: "mongo-az-a-1:27017", priority: 2 },  // 主 AZ 高优先级
  { _id: 1, host: "mongo-az-a-2:27017", priority: 2 },
  { _id: 2, host: "mongo-az-b-1:27017", priority: 1 },  // 备 AZ
  { _id: 3, host: "mongo-az-c-1:27017", priority: 0, votes: 0 }  // DR
]
rs.reconfig(cfg)

// 3. 监控告警
// - Primary 切换次数（异常切换告警）
// - 心跳延迟
// - 选举耗时