32. MongoDB 批量写入与吞吐优化

// insertMany：批量插入
db.orders.insertMany([
  { orderNo: "B-1", amount: 10 },
  { orderNo: "B-2", amount: 20 },
  { orderNo: "B-3", amount: 30 }
])

// bulkWrite：混合操作
db.orders.bulkWrite([
  { insertOne: { document: { orderNo: "B-4", amount: 40 } } },
  { updateOne: { filter: { orderNo: "B-1" }, update: { $inc: { amount: 1 } } } },
  { updateMany: { filter: { status: "stale" }, update: { $set: { status: "archived" } } } },
  { deleteOne: { filter: { orderNo: "B-2" } } }
])

// 批量请求的返回结果（示意）
db.orders.insertMany([{ orderNo: "B-5" }, { orderNo: "B-6" }])
// {
//   "acknowledged": true,
//   "insertedIds": { "0": ObjectId("..."), "1": ObjectId("...") }
// }

// 有序：遇到错误立即停止
db.orders.insertMany(
  [
    { orderNo: "O-1" },
    { orderNo: "O-1" },        // 违反唯一索引，触发错误
    { orderNo: "O-2" }         // 不会执行
  ],
  { ordered: true }
)

// 无序：跳过错误继续
db.orders.insertMany(
  [
    { orderNo: "U-1" },
    { orderNo: "U-1" },        // 报错但继续
    { orderNo: "U-2" }         // 会执行
  ],
  { ordered: false }
)

// 日志批量写入：无序 + 弱写关注，追求最大吞吐
db.app_events.insertMany(batchEvents, {
  ordered: false,
  writeConcern: { w: 1 }
})

// 批大小过大的后果：单条消息超过限制报错
db.orders.insertMany(bigBatch)
// 错误：BSON size limit exceeded 或 message exceeds 16MB

// 按体积估算批大小（示意）
const doc = { orderNo: "S-1", amount: 100, items: [] }
const docSize = Object.bsonsize(doc)          // 单条体积
const batchSize = Math.floor((8 * 1024 * 1024) / docSize)  // 留一半余量

// Node.js 驱动：批量插入并处理部分失败
const result = await orders.insertMany(batch, { ordered: false })
if (result.writeErrors && result.writeErrors.length > 0) {
  const failedIndexes = result.writeErrors.map(e => e.index)
  const retryBatch = failedIndexes.map(i => batch[i])
  // 过滤掉不可重试的错误（如违反唯一约束）后，再次批量提交
  await orders.insertMany(retryBatch, { ordered: false })
}

// 简单压测循环（mongosh 示意）
const docs = Array.from({ length: 10000 }, (_, i) => ({ seq: i, ts: new Date() }))
for (let size of [100, 500, 1000, 5000]) {
  const t0 = Date.now()
  let inserted = 0
  for (let i = 0; i < docs.length; i += size) {
    db.bench.insertMany(docs.slice(i, i + size), { ordered: false })
    inserted += Math.min(size, docs.length - i)
  }
  const ms = Date.now() - t0
  print(`batch=${size} total=${inserted} ops/s=${Math.round(inserted / (ms / 1000))}`)
}

// 同样的插入，索引越多越慢
db.orders.createIndex({ customerId: 1 })
db.orders.createIndex({ status: 1, createdAt: -1 })
db.orders.createIndex({ sku: 1 })
db.orders.createIndex({ warehouse: 1, zone: 1 })

// 每插入一条，要维护 1 个主键索引 + 4 个二级索引 = 5 次索引写入

// 写密集集合的最小索引集（示意）
db.app_events.createIndex({ deviceId: 1, ts: -1 })   // 查询所需
// 避免：为每个字段单独建索引

// 批量 + 弱确认：吞吐优先
db.logs.insertMany(batch, { ordered: false, writeConcern: { w: 1 } })

// 批量 + 强确认：一致性优先
db.orders.insertMany(batch, { ordered: false, writeConcern: { w: "majority" } })

// hashed 分片键：写入均匀打散到全部分片
sh.shardCollection("shop.orders", { orderId: "hashed" })

// ranged 分片键：写入可能集中于一个分片（热点）
sh.shardCollection("shop.orders", { createdAt: 1 })

// 查看写入相关指标
db.serverStatus().metrics
// {
//   "insert": { "total": 123456, "ops": 1234 },
//   "update": { "total": ..., "ops": ... },
//   "commands": { ... }
// }

// 查看当前正在执行的写入与阻塞
db.currentOp({ "command.insert": { $exists: true } })

// 对照压测：固定 5 万条数据，对比不同写关注
function benchWriteConcern(wc) {
  const docs = Array.from({ length: 50000 }, (_, i) => ({ seq: i }))
  const t0 = Date.now()
  for (let i = 0; i < docs.length; i += 1000) {
    db.bench.insertMany(docs.slice(i, i + 1000), { ordered: false, writeConcern: wc })
  }
  const ops = 50000 / ((Date.now() - t0) / 1000)
  print(`wc=${wc.w} ops/s=${Math.round(ops)}`)
  return Math.round(ops)
}
benchWriteConcern({ w: 1 })
benchWriteConcern({ w: "majority" })