MongoDB 教程 | 语法速查表

入门

连接到 MongoDB

mongosh 是现代的 MongoDB Shell(取代了旧版 mongo shell)。连接字符串遵循 URI 格式 mongodb://[user:pass@]host[:port]/[db][?options]。/admin 数据库是默认的认证数据库。你可以通过在 URI 末尾追加数据库名直接连接到特定数据库。

mongodb

# connect to a local server on default port
mongosh

# connect to a specific host and port
mongosh "mongodb://localhost:27017"

# connect with authentication
mongosh "mongodb://user:pass@localhost:27017/admin"

# connect to a specific database directly
mongosh "mongodb://localhost:27017/mydb"

# show connection info
db.getMongo().getDBNames()

数据库基础

MongoDB 惰性创建数据库和集合——只有在插入第一个文档后它们才会出现。'use mydb' 切换上下文,但在写入数据前不会创建数据库。show dbs 不会列出空数据库。dropDatabase 会永久删除所有集合及其数据。

mongodb

# show all databases
show dbs

# switch to (or create) a database
use mydb

# show the current database
db

# create is implicit: inserting into a
# non-existent collection creates both the
# collection and the database

# drop the current database
db.dropDatabase()

集合

集合类似于 SQL 表,用于存放文档。它们在首次插入时自动创建,但 createCollection 允许你设置选项,如 capped(固定大小集合,会覆盖旧文档——适用于日志)。stats() 显示存储大小、文档数量和索引详情。

mongodb

# list collections in current database
show collections
db.getCollectionNames()

# create a collection explicitly with options
db.createCollection("users", {
  capped: true,
  size: 5242880,
  max: 5000
})

# rename a collection
db.users.renameCollection("accounts")

# drop a collection
db.users.drop()

# get collection stats
db.users.stats()

BSON 数据类型

BSON 扩展了 JSON,增加了应用程序必需的额外类型:ObjectId(12 字节唯一 ID)、ISODate、NumberDecimal(货币精确小数——避免浮点舍入)、NumberLong(64 位整数)、BinData(二进制)等。对货币使用 NumberDecimal 可避免经典的浮点误差。

mongodb

# MongoDB stores BSON (binary JSON) with rich types
{
  _id: ObjectId("65a1b2c3d4e5f6a7b8c9d0e1"),
  name: "Alice",                       // String
  age: 30,                             // Int32
  score: NumberLong(9007199254740992), // Int64
  price: NumberDecimal("19.99"),       // Decimal128 (exact money)
  active: true,                        // Boolean
  birthday: ISODate("1995-08-15"),     // Date
  tags: ["red", "blue"],               // Array
  meta: { views: 100 },                // Embedded document
  data: BinData(0, "aGVsbG8="),        // Binary
  id: UUID("...")                      // UUID
}

Shell 辅助命令与帮助

mongosh 支持完整的 JavaScript,因此你可以在 shell 中编写循环、函数和变量。.pretty() 格式化输出以提高可读性(mongosh 默认启用)。load() 在 shell 上下文中运行 .js 文件——适用于重复的管理脚本。db.collection.help() 列出所有可用方法。

mongodb

# get help on database methods
db.help()

# get help on collection methods
db.users.help()

# list common shell commands
help

# pretty-print query results
db.users.find().pretty()

# print JSON in a compact form
db.users.find().toArray()

# execute a JavaScript file
mongosh --file script.js
load("script.js")

CRUD 操作

插入文档

如果你不提供 _id,MongoDB 会自动生成一个(ObjectId)。_id 在集合内必须唯一——重复的 _id 会引发写入错误。ordered:false 让 insertMany 在出错后继续插入剩余文档,提高批量加载的吞吐量。

mongodb

# insert a single document
db.users.insertOne({
  name: "Alice",
  email: "[email protected]",
  age: 30
})

# insert multiple documents
db.users.insertMany([
  { name: "Bob", age: 25 },
  { name: "Carol", age: 28 }
])

# insert with a custom _id
db.users.insertOne({ _id: 1, name: "Dave" })

# ordered insert (stops on error) vs unordered
db.users.insertMany(docs, { ordered: false })

查询文档

find() 返回一个游标(惰性求值);findOne() 返回单个文档或 null。投影控制返回哪些字段——不能混用包含和排除(_id 除外)。用 1 表示包含,0 表示排除。countDocuments 是精确的;旧版 count() 对带过滤的计数已弃用。

mongodb

# find all documents
db.users.find()

# find with a filter
db.users.find({ age: 30 })
db.users.find({ "address.city": "NYC" })

# find one matching document
db.users.findOne({ name: "Alice" })

# projection: include/exclude fields
db.users.find({}, { name: 1, age: 1, _id: 0 })
db.users.find({}, { email: 0 })

# count matching documents
db.users.countDocuments({ age: { $gte: 18 } })

更新文档

updateOne/updateMany 使用更新操作符($set、$inc 等)修改特定字段。replaceOne 替换整个文档(保留 _id)。upsert:true 在没有匹配时创建新文档——适用于'创建或更新'模式。始终使用 $set 更新字段;裸对象会替换文档(旧行为)。

mongodb

# update one document
db.users.updateOne(
  { name: "Alice" },
  { $set: { age: 31, status: "active" } }
)

# update many documents
db.users.updateMany(
  { status: "pending" },
  { $set: { status: "active" } }
)

# replace an entire document (keeps _id)
db.users.replaceOne(
  { name: "Alice" },
  { name: "Alice", age: 32, city: "NYC" }
)

# upsert: insert if no match found
db.users.updateOne(
  { name: "Eve" },
  { $set: { age: 22 } },
  { upsert: true }
)

更新操作符

更新操作符在文档级别原子地修改字段。$inc 是并发安全的(没有读-改-写竞争)。$min/$max 仅在新值更小/更大时更新。$currentDate 适用于'最后修改'时间戳。多个操作符可以在一次更新中组合使用。

mongodb

# $set and $unset
db.users.updateOne({n:"A"}, { $set: {age:30}, $unset: {temp:""} })

# $inc increments a numeric field
db.users.updateOne({n:"A"}, { $inc: {views: 1} })
db.users.updateOne({n:"A"}, { $inc: {balance: -50} })

# $rename a field
db.users.updateOne({n:"A"}, { $rename: {n: "name"} })

# $min / $max keep the smaller/larger value
db.users.updateOne({n:"A"}, { $min: {lowScore: 80} })

# $mul multiply the field value
db.users.updateOne({n:"A"}, { $mul: {price: 1.1} })

# $currentDate sets to current date
db.users.updateOne({n:"A"}, { $currentDate: {updatedAt: true} })

删除文档

deleteOne 删除第一个匹配项;deleteMany 删除所有匹配项。deleteMany({}) 清空集合但保留索引(如果你想保留索引定义,比 drop+recreate 更快)。findOneAndDelete 原子地返回被删除的文档。删除是不可逆的——务必仔细过滤。

mongodb

# delete one matching document
db.users.deleteOne({ name: "Alice" })

# delete all matching documents
db.users.deleteMany({ status: "inactive" })

# delete all documents in a collection
db.users.deleteMany({})

# findAndDelete: return the deleted document
db.users.findOneAndDelete({ name: "Bob" })

# remove is deprecated, prefer deleteOne/deleteMany
db.users.remove({ name: "Bob" }, { justOne: true })

数组更新操作符

$push 追加(允许重复);$addToSet 去重。$pull 删除匹配条件的元素。位置运算符 ($) 引用查询匹配的第一个数组元素——对于在不知道索引的情况下更新特定数组元素至关重要。$[] 更新所有数组元素。

mongodb

# $push append to an array
db.users.updateOne({n:"A"}, { $push: {tags: "new"} })

# $push with $each for multiple values
db.users.updateOne({n:"A"}, { $push: {tags: { $each: ["x","y"] } } })

# $addToSet adds only if not already present
db.users.updateOne({n:"A"}, { $addToSet: {tags: "x"} })

# $pull remove all matching array elements
db.users.updateOne({n:"A"}, { $pull: {tags: "x"} })

# $pop remove first (-1) or last (1) element
db.users.updateOne({n:"A"}, { $pop: {tags: 1} })

# update an array element by position
db.users.updateOne({}, { $set: {"tags.0": "first"} })
db.users.updateOne({"tags":"old"}, { $set: {"tags.$":"new"} })

查询文档

比较操作符

比较操作符是查询的基础。$in 比同一字段上多个 $or 条件高效得多。你可以在一个字段上组合多个操作符(例如 $gte 和 $lte 表示范围)。$exists:true 查找包含该字段的文档,$exists:false 查找不包含的文档。

mongodb

# $eq equal (same as direct value)
db.products.find({ price: { $eq: 100 } })

# $ne not equal
db.products.find({ price: { $ne: 100 } })

# $gt, $gte, $lt, $lte
db.products.find({ price: { $gte: 50, $lte: 200 } })

# $in matches any value in an array
db.products.find({ category: { $in: ["books", "music"] } })

# $nin not in array
db.products.find({ category: { $nin: ["books"] } })

# $exists check field presence
db.products.find({ discount: { $exists: true } })

逻辑操作符

隐式 AND(过滤器中的逗号)是最常见且高效的。只有当需要同一字段上的多个条件时才使用显式 $and(同一字段上的隐式 AND 会覆盖前面的条件)。$or 使用独立的索引扫描并合并结果——确保 $or 中的字段已索引以提高性能。

mongodb

# $and (explicit, useful for same-field conditions)
db.users.find({ $and: [
  { age: { $gte: 18 } },
  { age: { $lte: 65 } }
]})

# $or matches if ANY condition is true
db.users.find({ $or: [
  { status: "active" },
  { vip: true }
]})

# $nor matches if NO condition is true
db.users.find({ $nor: [{ status: "active" }]})

# $not inverts a condition
db.users.find({ age: { $not: { $gt: 18 } } })

# implicit AND (comma-separated)
db.users.find({ age: { $gte: 18 }, status: "active" })

元素与求值操作符

$type 按 BSON 类型过滤(适用于混合类型字段)。$regex 支持 'i'(不区分大小写)等选项——但没有前缀锚点的正则表达式无法高效使用索引。$expr 支持文档字段之间的比较,普通操作符做不到。谨慎使用 $expr,因为它可能绕过索引。

mongodb

# $type filter by BSON data type
db.users.find({ age: { $type: "int" } })
db.users.find({ age: { $type: ["int", "double"] } })

# $regex regular expression match
db.users.find({ name: { $regex: "^Al", $options: "i" } })

# $mod modulo operation
db.users.find({ age: { $mod: [2, 0] } })  # even ages

# $expr compare fields within a document
db.orders.find({ $expr: { $gt: ["$total", "$budget"] } })

# $jsonSchema validate against a schema
db.users.find({ $jsonSchema: {
  required: ["name", "email"]
}})

数组查询操作符

$all 查找包含所有指定值的数组(顺序无关)。当一个数组元素必须同时满足多个条件时,$elemMatch 至关重要(普通查询会分别匹配不同元素)。$size 仅匹配精确长度——没有带范围的 $size;改为预计算并将长度存储为单独字段。

mongodb

# $all: array contains ALL specified values
db.posts.find({ tags: { $all: ["mongo", "db"] } })

# $elemMatch: at least one element matches all conditions
db.students.find({ scores: { $elemMatch: {
  subject: "math", score: { $gte: 90 }
}}})

# $size: array of exact length
db.users.find({ tags: { $size: 3 } })

# match if array contains the value
db.posts.find({ tags: "mongo" })

# match a specific array index
db.users.find({ "scores.0": 95 })  # first score is 95

游标方法与排序

skip+limit 实现分页,但对于大偏移量 skip 会变慢(它会扫描所有跳过的文档)。对于深度分页,使用键集分页:用排序字段大于最后看到的值进行查询过滤。sort() 可以使用索引来避免内存排序——对于大集合务必索引排序字段。

mongodb

# limit and skip (pagination)
db.users.find().limit(10)
db.users.find().skip(20).limit(10)  # page 3

# sort: 1 ascending, -1 descending
db.users.find().sort({ age: 1, name: -1 })

# count results
db.users.find({active:true}).count()

# iterate the cursor in the shell
db.users.find().forEach(doc => print(doc.name))

# get a specific field's distinct values
db.users.distinct("city")

索引

创建索引

索引极大地加速查询,但会减慢写入并消耗磁盘。单字段索引支持该字段在任一排序方向上的查询。复合索引遵循 ESR(相等、排序、范围)规则以获得最佳顺序。TTL 索引会在一段时间后自动删除文档——非常适合会话和日志。

mongodb

# create a single-field index
db.users.createIndex({ email: 1 })          # ascending
db.users.createIndex({ age: -1 })            # descending

# create a unique index
db.users.createIndex({ email: 1 }, { unique: true })

# create a compound index
db.users.createIndex({ lastName: 1, firstName: 1 })

# create a text index
db.posts.createIndex({ title: "text", body: "text" })

# create a TTL (time-to-live) index
db.sessions.createIndex({ createdAt: 1 },
  { expireAfterSeconds: 3600 })

# name an index explicitly
db.users.createIndex({ email: 1 }, { name: "email_idx" })

查看与管理索引

getIndexes 列出所有索引及其键和选项。hideIndex/unhideIndex 让你测试删除索引的影响而无需实际删除——查询规划器会忽略隐藏的索引。如果性能保持不变,就可以安全删除。在索引变更前后始终运行 explain()。

mongodb

# list all indexes on a collection
db.users.getIndexes()

# total index size in bytes
db.users.totalIndexSize()

# drop an index by name
db.users.dropIndex("email_1")

# drop all indexes except _id
db.users.dropIndexes()

# hide an index (test impact before dropping)
db.users.hideIndex("email_1")
db.users.unhideIndex("email_1")

# check if queries use indexes
db.users.find({email:"[email protected]"}).explain("executionStats")

特殊索引类型

2dsphere 索引支持地理查询($near、$geoWithin)。哈希索引支持基于哈希的分片以实现均匀数据分布。通配符索引($**)覆盖文档中不可预测/可变的字段名——适用于多态数据。部分索引仅索引匹配的文档,在你只查询子集时节省空间。

mongodb

# geospatial 2dsphere index (GeoJSON points)
db.places.createIndex({ location: "2dsphere" })
db.places.find({
  location: { $near: {
    $geometry: { type: "Point", coordinates: [-73.99, 40.73] },
    $maxDistance: 1000
  }}
})

# hashed index (for sharding)
db.users.createIndex({ user_id: "hashed" })

# wildcard index (index arbitrary field names)
db.products.createIndex({ "$**": 1 })

# partial index (only matching documents)
db.users.createIndex({ email: 1 },
  { partialFilterExpression: { active: true } })

执行计划与查询计划

explain() 是主要的性能工具。COLLSCAN(集合扫描)表示未使用索引——对大集合来说是危险信号。IXSCAN 表示使用了索引。totalDocsExamined >> nReturned 表示索引不佳(检查很多文档却返回很少)。hint() 强制使用特定索引以测试规划器的选择。

mongodb

# explain a query (basic)
db.users.find({age:30}).explain()

# explain with execution stats (timing)
db.users.find({age:30}).explain("executionStats")

# key fields in the output:
#   winningPlan.stage    -> COLLSCAN (bad) or IXSCAN (good)
#   totalDocsExamined    -> should be close to nReturned
#   executionTimeMillis  -> query duration
#   indexUsed            -> which index was chosen

# force a specific index (testing)
db.users.find({age:30}).hint("age_1")
db.users.find({age:30}).hint({age:1})

索引最佳实践

ESR(相等、排序、范围)规则是最重要的复合索引设计原则:先放相等过滤字段,再放排序字段,最后放范围字段。覆盖查询(所有字段都在索引中)从不获取文档——最快的查询方式。避免使用未使用的索引;每个索引都会增加写入开销。

mongodb

# ESR rule: Equality, Sort, Range for compound indexes
# equality fields first, then sort, then range
db.orders.createIndex({
  status: 1,        # Equality  (filter)
  date: 1,          # Sort      (order)
  amount: 1         # Range     (comparison)
})

# avoid over-indexing: each index slows writes
db.orders.getIndexes().length

# covered query: index satisfies query without fetching doc
db.users.createIndex({ name: 1, email: 1 })
db.users.find({ name: "A" }, { _id: 0, name: 1, email: 1 })

# background index build (deprecated in 4.2+, builds don't block)
db.users.createIndex({ name: 1 })  # non-blocking by default

聚合框架

聚合管道基础

聚合管道按顺序通过阶段处理文档。管道早期的 $match 减少工作集(并可使用索引)。$group 类似 SQL GROUP BY。$count 是 $group + $project 的快捷方式。顺序很重要:先过滤,再分组,再排序,最后限制。

mongodb

# a pipeline is an array of stages
db.orders.aggregate([
  { $match: { status: "shipped" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } },
  { $limit: 10 }
])

# each stage transforms documents and passes them
# to the next stage

# count documents
db.orders.aggregate([
  { $match: { status: "shipped" } },
  { $count: "shippedCount" }
])

$match、$project 与 $limit

始终尽早放置 $match——它减少后续所有阶段的文档数量,并可以使用索引。$project 重塑文档并使用表达式计算新字段。在 $group 之前在 $project 中重命名或计算字段很常见。$unset 是排除字段的简写。

mongodb

# $match filters documents (like find)
{ $match: { status: "active", age: { $gte: 18 } } }

# $project reshapes documents (1 include, 0 exclude)
{ $project: { name: 1, _id: 0, "address.city": 1 } }

# $project with computed fields
{ $project: {
  fullName: { $concat: ["$first", " ", "$last"] },
  ageNextYear: { $add: ["$age", 1] }
}}

# $limit and $skip
{ $limit: 10 }
{ $skip: 20 }

# $unset removes fields (inverse of $project)
{ $unset: "tempField" }

$group 与累加器

$group 是核心聚合阶段。_id 指定分组键(使用 null 聚合所有文档)。$push 为每组构建值数组;$addToSet 去重。$first/$last 引用组内的文档顺序——如果顺序重要,请在 $group 之前使用 $sort。每阶段的内存限制为 100MB(大数据集使用 allowDiskUse)。

mongodb

# $group: group by a field and aggregate
{ $group: {
  _id: "$category",
  count: { $sum: 1 },
  totalRevenue: { $sum: "$amount" },
  avgPrice: { $avg: "$price" },
  minPrice: { $min: "$price" },
  maxPrice: { $max: "$price" },
  products: { $push: "$name" },     # array of values
  firstOrder: { $first: "$date" },
  lastOrder: { $last: "$date" }
}}

# group by multiple fields
{ $group: { _id: { category: "$cat", status: "$status" } } }

# group ALL documents together (_id: null)
{ $group: { _id: null, total: { $sum: "$amount" } } }

$lookup(连接)

$lookup 执行左外连接——类似 SQL LEFT JOIN。连接的数据放在由 'as' 命名的数组字段中。管道形式允许你过滤和转换连接的文档。$unwind 解构数组字段(每个数组元素一个输出文档)——常在 $lookup 之后用于展平结果。在 MongoDB 中,连接比内嵌数据效率低。

mongodb

# $lookup: join another collection (left outer join)
{ $lookup: {
  from: "orders",
  localField: "_id",
  foreignField: "customerId",
  as: "orders"
}}

# with pipeline (filtered/transformed join)
{ $lookup: {
  from: "orders",
  let: { custId: "$_id" },
  pipeline: [
    { $match: { $expr: { $eq: ["$customerId", "$$custId"] } } },
    { $project: { amount: 1, _id: 0 } }
  ],
  as: "recentOrders"
}}

# $unwind: flatten an array field into multiple docs
{ $unwind: "$orders" }

$unwind、$sort 与 $facet

$facet 在同一输入上并行运行多个子管道——非常适合需要多种数据聚合(Top N、计数、细分)的仪表板。$bucket 自动将值分类到范围中。带 preserveNullAndEmptyArrays 的 $unwind 保留数组为空或缺失的文档——有助于避免管道中的数据丢失。

mongodb

# $unwind with preserveNullAndEmptyArrays
{ $unwind: {
  path: "$tags",
  preserveNullAndEmptyArrays: true
}}

# $sort within pipeline
{ $sort: { total: -1 } }

# $facet: run multiple pipelines in parallel
{ $facet: {
  "topCustomers": [{ $sort: {total:-1} }, { $limit: 5 }],
  "byCategory":   [{ $group: { _id: "$cat" } }],
  "totalCount":   [{ $count: "n" }]
}}

# $bucket: group into ranges automatically
{ $bucket: {
  groupBy: "$price",
  boundaries: [0, 50, 100, 500],
  default: "other",
  output: { count: { $sum: 1 } }
}}

全文搜索与地理空间

全文搜索

文本索引通过 $text 启用全文搜索。单词被分词和词干化。搜索默认是 OR;短语需要转义引号;减号前缀排除。$meta:'textScore' 返回相关性;按其排序可获得最佳匹配。每个集合只能有一个文本索引。对于高级搜索,使用 Atlas Search(基于 Lucene)。

mongodb

# create a text index on multiple fields
db.posts.createIndex({
  title: "text",
  body: "text",
  tags: "text"
})

# search for words (OR by default)
db.posts.find({ $text: { $search: "mongodb tutorial" } })

# exact phrase (escape quotes)
db.posts.find({ $text: { $search: "\"mongo db\"" } })

# exclude a word (minus)
db.posts.find({ $text: { $search: "mongodb -sql" } })

# sort by relevance score
db.posts.find(
  { $text: { $search: "mongodb" } },
  { score: { $meta: "textScore" } }
).sort({ score: { $meta: "textScore" } })

地理空间 2dsphere

GeoJSON 坐标是 [经度, 纬度](x, y 顺序——常见陷阱)。2dsphere 索引支持 $near(按距离排序)、$geoWithin(在形状内)和 $geoIntersects。$maxDistance 以米为单位。距离计算使用球体模型。对于 2D 平面数据(罕见),改用 '2d' 索引。

mongodb

# store points as GeoJSON
db.places.insertOne({
  name: "Park",
  location: {
    type: "Point",
    coordinates: [-73.99, 40.73]   # [lng, lat]
  }
})

# create a 2dsphere index
db.places.createIndex({ location: "2dsphere" })

# find nearby places within a distance (meters)
db.places.find({
  location: { $near: {
    $geometry: { type: "Point", coordinates: [-73.99, 40.73] },
    $maxDistance: 1000
  }}
})

# find places within a polygon
db.places.find({
  location: { $geoWithin: {
    $geometry: { type: "Polygon", coordinates: [[[-73,40],[-74,40],[-74,41],[-73,41],[-73,40]]] }
  }}
})

地理空间距离与操作符

$nearSphere 使用精确的球面几何(更适合大距离)。$geoNear 聚合阶段是唯一能在结果中返回计算距离的方式——distanceField 以米(球面)或弧度存储。$centerSphere 对半径使用弧度(弧度 = 距离_公里 / 6371)。

mongodb

# $nearSphere: accurate spherical distance
db.places.find({
  location: { $nearSphere: {
    $geometry: { type: "Point", coordinates: [-73.99, 40.73] },
    $minDistance: 100,
    $maxDistance: 5000
  }}
})

# $geoNear aggregation stage (includes distance)
db.places.aggregate([{
  $geoNear: {
    near: { type: "Point", coordinates: [-73.99, 40.73] },
    distanceField: "dist",
    maxDistance: 2000,
    spherical: true
  }
}])

# $center / $centerSphere for simple radius
db.places.find({
  location: { $geoWithin: { $centerSphere: [[-73.99,40.73], 0.01] } }
})

正则表达式与模式匹配

前缀锚定正则表达式(/^Ali/)可以高效使用索引;未锚定的模式(/ali/)不能使用索引并扫描整个集合。对于大集合上的不区分大小写搜索,优先使用文本索引或 Atlas Search 而非正则表达式。$regexMatch(聚合)允许你在计算字段和条件中使用正则表达式结果。

mongodb

# case-insensitive regex
db.users.find({ name: { $regex: /alice/i } })

# prefix-anchored regex CAN use an index
db.users.find({ name: { $regex: /^Ali/ } })

# wildcard match
db.users.find({ email: { $regex: /@example\.com$/ } })

# $regex with options string
db.users.find({ name: { $regex: "ali", $options: "ix" } })

# use $expr with $regexMatch for conditional logic
db.users.aggregate([{
  $project: {
    name: 1,
    isAlice: { $regexMatch: { input: "$name", regex: /^ali/i } }
  }
}])

通配符与灵活查询

$where 对每个文档运行 JavaScript——非常慢,应避免;$expr 通常可以替代它。$elemMatch 确保一个数组元素匹配所有条件(普通逗号查询会匹配不同元素)。$expr 支持普通操作符无法做到的字段间比较——但它可能无法高效使用索引。

mongodb

# $jsonSchema validation as a query filter
db.users.find({ $jsonSchema: {
  bsonType: "object",
  required: ["name"],
  properties: { age: { bsonType: "int", minimum: 0 } }
}})

# $where (JavaScript) — slow, avoids indexes
db.users.find({ $where: "this.a + this.b > 10" })

# $mod modulo operator
db.users.find({ age: { $mod: [10, 0] } })

# $expr: compare two fields
db.orders.find({ $expr: { $lt: ["$paid", "$total"] } })

# $elemMatch with multiple conditions on array elements
db.scores.find({ results: {
  $elemMatch: { $gte: 80, $lt: 90 }
}})

事务

ACID 事务

多文档事务提供跨集合的 ACID 保证(需要副本集或分片集群)。将 session 传递给事务中的每个操作。commitTransaction 使更改永久且可见;abortTransaction 回滚所有更改。保持事务简短——它们持有锁并可能影响性能。

mongodb

# multi-document transaction (replica set required)
const session = db.getMongo().startSession()
session.startTransaction()

try {
  db.accounts.updateOne(
    { name: "Alice" }, { $inc: { balance: -100 } }, { session }
  )
  db.accounts.updateOne(
    { name: "Bob" }, { $inc: { balance: 100 } }, { session }
  )
  session.commitTransaction()
} catch (e) {
  session.abortTransaction()
  print("Transaction aborted: " + e)
} finally {
  session.endSession()
}

事务读写关注

读关注控制读取的'新鲜度'和一致性。'snapshot' 提供跨分片的事务一致性视图——适用于报表。写关注控制持久性:w:majority 确保写入在确认前已复制(能在主节点故障转移中存活);j:true 等待日志刷新。更高的关注级别更安全但更慢。

mongodb

# set transaction-level read/write concern
session.startTransaction({
  readConcern: { level: "snapshot" },
  writeConcern: { w: "majority", j: true, wtimeout: 5000 }
})

# read concern levels:
#   "local"      - sees latest local data (default)
#   "majority"   - committed, replicated to majority
#   "snapshot"   - consistent snapshot across shards

# write concern levels:
#   { w: 1 }             - ack from primary (default)
#   { w: "majority" }    - ack from majority of members
#   { j: true }          - wait for journal write

重试逻辑与错误

MongoDB 用 errorLabels 标记可重试错误:TransientTransactionError(重试整个事务)和 UnknownCommitResult(仅重试提交)。生产驱动程序提供内置重试。尽可能设计幂等事务。避免运行时间长、持有锁并阻塞其他操作的事务。

mongodb

# transactions can fail with TransientTransactionError
# retry the whole transaction on these errors

function runTransaction(txnFunc) {
  while (true) {
    const session = db.getMongo().startSession()
    try {
      session.startTransaction()
      txnFunc(session)
      session.commitTransaction()
      return
    } catch (e) {
      session.abortTransaction()
      if (e.errorLabels &&
          e.errorLabels.includes("TransientTransactionError")) {
        continue  // retry
      }
      throw e   // non-retryable error
    } finally {
      session.endSession()
    }
  }
}

原子单文档操作

单文档操作默认是原子的——不需要事务。findOneAndUpdate/findOneAndReplace/findOneAndDelete 原子地返回文档(非常适合生成顺序 ID)。对于许多工作负载,将相关数据内嵌在一个文档中并使用原子操作符($inc、$push)完全避免了多文档事务的需要。

mongodb

# single-document operations are ALWAYS atomic
db.counters.updateOne(
  { _id: "pageViews" },
  { $inc: { count: 1 } }
)

# findAndModify: atomically find & update, return the doc
db.sequences.findOneAndUpdate(
  { _id: "orderId" },
  { $inc: { seq: 1 } },
  { returnNewDocument: true }
)

# atomic upsert (create-or-increment)
db.counters.updateOne(
  { _id: "visits", date: "2025-01-01" },
  { $inc: { count: 1 } },
  { upsert: true }
)

# use $expr for conditional atomic updates
db.products.updateMany(
  { $expr: { $lt: ["$stock", "$reorderLevel"] } },
  { $set: { needsRestock: true } }
)

两阶段提交(模式)

两阶段提交模式在没有原生多文档事务的情况下模拟分布式事务。它跟踪'pending'状态,以便恢复进程在传输中途崩溃时可以恢复或回滚。当有原生事务可用时请改用原生事务——此模式复杂且容易出错,主要出于教育和遗留目的保留。

mongodb

# pattern for atomic transfers without transactions
# (pre-4.0 compatibility, or cross-system transfers)

# 1. create a pending transaction document
db.transactions.insertOne({
  _id: "t1", from: "Alice", to: "Bob", amount: 100, state: "pending"
})

# 2. debit source
db.accounts.updateOne(
  { name: "Alice", pendingTransactions: { $ne: "t1" } },
  { $inc: { balance: -100 }, $push: { pendingTransactions: "t1" } }
)

# 3. credit destination
db.accounts.updateOne(
  { name: "Bob", pendingTransactions: { $ne: "t1" } },
  { $inc: { balance: 100 }, $push: { pendingTransactions: "t1" } }
)

# 4. mark transaction done
db.transactions.updateOne({ _id: "t1" }, { $set: { state: "done" } })

# 5. clean up pending references
db.accounts.updateMany({},
  { $pull: { pendingTransactions: "t1" } })

数据建模

内嵌与引用

MongoDB 建模的关键决策:对于一起访问且很少更改的数据内嵌;对于大型、共享或独立更新的数据引用。当'has-a'关系是独占的(用户的地址)时内嵌。当数据被重用(多个订单中的产品)或无界增长(用户数百万条日志条目)时引用。

mongodb

# EMBED: store related data in one document
db.users.insertOne({
  name: "Alice",
  addresses: [
    { city: "NYC", zip: "10001" },
    { city: "LA", zip: "90001" }
  ]
})
# Pros: single read, atomic update
# Cons: document size limit (16MB), data duplication

# REFERENCE: store IDs, join with $lookup
db.users.insertOne({ name: "Alice" })
db.orders.insertOne({
  customerId: ObjectId("..."),
  amount: 100
})
# Pros: no duplication, independent updates
# Cons: needs joins, more queries

一对多关系

对于'少'的关系,内嵌。对于'多',从子文档引用(每个子文档上的父 ID)。对于'非常多'(数百万),使用桶模式:将子文档分组到父级拥有的桶中(例如按月),以平衡文档大小和查询速度。这避免了无界数组和过多的子文档。

mongodb

# FEW: embed (e.g., a few addresses per user)
db.users.insertOne({
  name: "Alice",
  addresses: [{ city: "NYC" }, { city: "LA" }]
})

# MANY: reference with parent pointer
db.posts.insertOne({ title: "Hello", authorId: ObjectId("...") })

# VERY MANY: reference from parent (bucket pattern)
db.users.updateOne(
  { _id: userId },
  { $push: { postIds: postId } }   # beware 16MB limit
)

# BUCKET pattern for time series (thousands of items)
db.buckets.insertOne({
  ownerId: userId,
  date: "2025-01",
  measurements: [ {t: 1, v: 30}, {t: 2, v: 31} ]
})

多对多关系

对于多对多,根据查询模式和数据大小选择方法。双向数组速度快但需要双重维护。连接集合(映射表)最灵活且扩展性最好——它可以存储关系元数据(注册日期、成绩)并避免膨胀父文档。

mongodb

# option 1: arrays of ObjectIds on both sides
db.students.insertOne({ name: "Alice", courseIds: [c1, c2] })
db.courses.insertOne({ name: "Math", studentIds: [s1, s2] })

# option 2: one side references (smaller side)
db.students.insertOne({ name: "Alice", courseIds: [c1, c2] })

# option 3: mapping/junction collection
db.enrollments.insertOne({
  studentId: s1,
  courseId: c1,
  enrolledAt: ISODate(),
  grade: "A"
})

# query: find a student's courses
db.enrollments.find({ studentId: s1 })

模式设计模式

属性模式处理具有可变属性的实体(不同规格的产品)——避免稀疏字段并支持动态属性。多态模式使用类型鉴别器在一个集合中存储相关但不同的形状。异常模式通过引用而非内嵌来处理特殊文档(拥有数百万粉丝的名人)。

mongodb

# ATTRIBUTE PATTERN: flexible attributes (e.g., products)
db.products.insertOne({
  name: "Laptop",
  attributes: [
    { k: "cpu", v: "i7" },
    { k: "ram", v: 16 }
  ]
})
# query: db.products.find({ "attributes.k": "cpu", "attributes.v": "i7" })

# POLYMORPHIC PATTERN: one collection, type discriminator
db.events.insertMany([
  { type: "click", url: "/home", userId: 1 },
  { type: "purchase", orderId: 99, total: 50 }
])

# OUTLIER PATTERN: cap large lists
db.users.updateOne({n:"A"}, { $set: { reviews: "see reviews collection" } })

模式验证

模式验证在写入时强制执行文档结构——MongoDB 对固定模式的回应。validationLevel 'strict' 验证所有写入;'moderate' 仅验证插入和完全更新。validationAction 'error' 拒绝无效文档;'warn' 记录但接受。验证补充(而非取代)应用层验证。

mongodb

# add a JSON schema validator to a collection
db.createCollection("users", {
  validator: {
    $jsonSchema: {
      bsonType: "object",
      required: ["name", "email"],
      properties: {
        name: { bsonType: "string", maxLength: 100 },
        email: { bsonType: "string", pattern: "^\S+@\S+$" },
        age: { bsonType: "int", minimum: 0, maximum: 150 },
        role: { enum: ["admin", "user", "guest"] }
      }
    }
  },
  validationLevel: "strict",
  validationAction: "error"
})

# add validation to an existing collection
db.runCommand({ collMod: "users", validator: { ... } })

# check existing validation rules
db.getCollectionInfos({ name: "users" })

副本集

副本集基础

副本集提供高可用性和读取扩展。只有一个成员是主节点(接受写入);其他成员异步复制。如果主节点故障,选举会自动提升一个从节点。奇数投票成员(最少 3 个)可避免选举平局。rs.status() 显示每个成员的健康状况、状态和复制延迟。

mongodb

# a replica set = 1 primary + N secondaries
# all writes go to the primary; secondaries replicate

# initiate a replica set
rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "host1:27017" },
    { _id: 1, host: "host2:27017" },
    { _id: 2, host: "host3:27017" }
  ]
})

# check replica set status
rs.status()
rs.conf()

# step down the primary (force an election)
rs.stepDown(60)

# add a new member
rs.add("host4:27017")
rs.remove("host4:27017")

读偏好

读偏好平衡一致性和负载。primary 保证你总是读取最新写入。secondary/secondaryPreferred 从主节点卸载读取,但数据可能过时(复制延迟,通常为毫秒级)。nearest 最小化延迟——适合地理分布的应用。分析/报表工作负载应使用从节点读取。

mongodb

# control where reads are routed
# primary (default) - always read from primary (strongest consistency)
db.users.find().readPref("primary")

# primaryPreferred - primary if available, else secondary
db.users.find().readPref("primaryPreferred")

# secondary - only secondaries (may be stale)
db.users.find().readPref("secondary")

# secondaryPreferred - prefer secondaries to offload primary
db.users.find().readPref("secondaryPreferred")

# nearest - lowest latency member
db.users.find().readPref("nearest")

# set in connection string
mongodb://host1:27017/mydb?replicaSet=rs0&readPreference=secondary

写关注

写关注用速度换取安全性。w:1 在主节点写入后立即确认(快但如果主节点在复制前崩溃则有数据丢失风险)。w:majority 确保写入已复制——能在单节点故障中存活(推荐用于重要数据)。j:true 等待磁盘日志刷新。在集群级别设置合理的默认值。

mongodb

# control how durable a write must be before ack
# w: 1 - primary acks (default, fastest, least safe)
db.users.insertOne(doc, { writeConcern: { w: 1 } })

# w: "majority" - majority of members confirmed
db.users.insertOne(doc, { writeConcern: { w: "majority" } })

# j: true - wait for journal flush (survives crash)
db.users.insertOne(doc, { writeConcern: { w: 1, j: true } })

# wtimeout - max wait time in ms
db.users.insertOne(doc, {
  writeConcern: { w: "majority", wtimeout: 5000 }
})

# set default write concern for the cluster
db.adminCommand({
  setDefaultRWConcern: { defaultWriteConcern: { w: "majority" } }
})

仲裁者与选举

仲裁者提供一票以达到多数但不存储数据——适用于双节点部署,但第三个数据节点更可取。Priority 0 成员永远不会成为主节点(适合专用分析或灾难恢复节点)。隐藏成员被排除在读偏好 'nearest' 之外,适合备份。延迟成员滞后,可从意外删除中恢复。

mongodb

# an arbiter votes in elections but holds no data
rs.addArb("host5:27017")

# useful to break ties in even-member sets (e.g., 2 data + 1 arbiter)
# but prefer 3 data members for durability

# force a member to never become primary (priority 0)
rs.conf().members[1].priority = 0
rs.reconfig(rs.conf())

# hidden member: not visible to clients, used for analytics/backup
rs.conf().members[2].hidden = true
rs.conf().members[2].priority = 0
rs.reconfig(rs.conf())

# delayed member: lags behind (insurance against human errors)
rs.conf().members[3].secondaryDelaySecs = 3600
rs.conf().members[3].priority = 0

Oplog 与复制内部机制

oplog(操作日志)是一个记录每次写入的固定集合——从节点按顺序应用这些条目来复制。如果从节点落后超过 oplog 窗口,它需要完全重新同步。printReplicationInfo 显示 oplog 的时间跨度;printSecondaryReplicationInfo 显示每个从节点的延迟。更大的 oplog 可适应更长的停机时间。

mongodb

# the oplog is a capped collection of all writes
use local
db.oplog.rs.stats()
db.oplog.rs.find().sort({ $natural: -1 }).limit(5)

# each oplog entry is an idempotent operation
# { ts: ..., op: "i", ns: "mydb.users", o: { ... } }

# check replication lag
rs.printReplicationInfo()   # primary's oplog window
rs.printSecondaryReplicationInfo()  # each secondary's lag

# resync a stale secondary (full re-sync)
rs.syncFrom("host2:27017")

# change the oplog size (MongoDB 4.4+)
db.adminCommand({ replSetResizeOplog: 1, size: 16384 })  # MB

分片

分片基础

分片将集合水平分布到多个分片上。分片键决定数据分布,在分片后不可变。哈希分片均匀分布(适合 ObjectId 等单调递增键);范围分片支持定向查询但可能导致热点。仔细选择分片键——之后不容易更改。

mongodb

# sharding distributes data across multiple machines
# requires: config servers + shard replica sets + mongos router

# enable sharding on a database
sh.enableSharding("mydb")

# shard a collection (index on shard key required first)
sh.shardCollection("mydb.users", { userId: 1 })

# hash-based sharding (even distribution)
db.users.createIndex({ userId: "hashed" })
sh.shardCollection("mydb.users", { userId: "hashed" })

# view shard status
sh.status()
sh.addShard("rs1/host1:27017,host2:27017")

# balancer moves chunks between shards automatically
sh.startBalancer()
sh.stopBalancer()

分片键选择

分片键是最关键的分片决策。它必须支持你的常见查询(没有分片键的查询会命中所有分片——散射-聚集)。高基数防止不可移动的巨型块。单调键(时间戳、ObjectId)导致所有插入命中一个分片——对这些使用哈希分片。复合键应以最具选择性的字段开头。

mongodb

# GOOD shard key properties:
# - high cardinality (many distinct values)
# - low frequency (no single value dominates)
# - non-monotonic (avoids hot shards) OR use hashed

# hashed shard key (even distribution)
sh.shardCollection("mydb.logs", { _id: "hashed" })

# ranged shard key (enables targeted queries)
sh.shardCollection("mydb.events", { region: 1, date: 1 })

# compound shard key
sh.shardCollection("mydb.orders", { customerId: 1, orderId: 1 })

# check chunk distribution
db.getSiblingDB("config").chunks.find({ ns: "mydb.users" })

# see data distribution per shard
db.users.getShardDistribution()

区域与标签分片

区域(标签分片)将特定数据范围路由到特定分片——实现数据驻留合规(GDPR 要求欧盟数据在欧盟)、地理位置(快速区域读取)或硬件分层(热数据在 SSD,冷数据在 HDD)。均衡器在迁移块时遵守区域范围。先定义分片标签,再定义标签范围。

mongodb

# tag shards by region (data locality / compliance)
sh.addShardTag("shard-us-east", "US")
sh.addShardTag("shard-eu", "EU")

# associate a key range with a tag
sh.addTagRange(
  "mydb.users",
  { region: "US", _id: MinKey },
  { region: "US", _id: MaxKey },
  "US"
)
sh.addTagRange(
  "mydb.users",
  { region: "EU", _id: MinKey },
  { region: "EU", _id: MaxKey },
  "EU"
)

# the balancer migrates tagged data to tagged shards
# useful for data residency (GDPR) and geo-locality

块管理

块是分片间数据移动的单位。均衡器自动拆分和迁移块以保持分片平衡。巨型块超过大小限制无法迁移——用好的分片键基数来避免。默认 128MB 块大小平衡了迁移速度和开销。手动块操作用于故障排除,不是常规操作。

mongodb

# chunks are ranges of shard key values
# default chunk size: 128MB (configurable)

# view chunks for a collection
db.getSiblingDB("config").chunks.find({ ns: "mydb.users" })

# manually move a chunk (testing)
sh.moveChunk("mydb.users", { userId: 100 }, "shard2")

# split a chunk at a specific value
sh.splitAt("mydb.users", { userId: 1000 })

# merge contiguous chunks
sh.mergeChunks("mydb.users",
  { userId: MinKey },
  { userId: 1000 }
)

# find chunks too large to migrate (jumbo chunks)
sh.status(true)   # verbose

分片集群组件

分片集群有三个组件:配置服务器(元数据,部署为副本集)、mongos 路由器(客户端连接的无状态查询路由器)和分片(每个都是持有数据子集的副本集)。客户端从不直接连接分片。移除分片会触发数据迁移到剩余分片——这是一个应被监控的缓慢过程。

mongodb

# config servers store metadata and routing info
# (deploy as a 3-member replica set)
# config server connection string:
# mongodb://cfg1:27019,cfg2:27019,cfg3:27019/config?replicaSet=cfgRS

# mongos is the query router (stateless, scale horizontally)
# clients connect to mongos, not shards directly
mongos --configdb cfgRS/cfg1:27019,cfg2:27019,cfg3:27019

# each shard is a replica set
sh.addShard("shardRS1/s1a:27017,s1b:27017,s1c:27017")
sh.addShard("shardRS2/s2a:27017,s2b:27017,s2c:27017")

# check cluster status
sh.status()

# remove a shard (migrates its data first)
db.adminCommand({ removeShard: "shardRS2" })

安全与认证

认证与用户

认证验证身份。在启用认证之前创建管理员用户,否则会被锁在外面。内置角色从只读到完全管理员。在 shell 中使用 passwordPrompt() 以避免历史记录中的明文密码。每个用户限定于一个数据库,但可以在其他数据库上有角色。在生产环境中启用 TLS(net.tls.mode)。

mongodb

# enable auth: start mongod with --auth
# or set security.authorization: "enabled" in config

# create the first admin user (before enabling auth)
use admin
db.createUser({
  user: "admin",
  pwd: passwordPrompt(),   // prompts securely
  roles: [ { role: "userAdminAnyDatabase", db: "admin" },
           { role: "readWriteAnyDatabase", db: "admin" } ]
})

# create a user for a specific database
use mydb
db.createUser({
  user: "appUser",
  pwd: passwordPrompt(),
  roles: [ { role: "readWrite", db: "mydb" } ]
})

# authenticate as a user
db.auth("appUser", "password")

内置角色与自定义角色

内置角色涵盖常见情况;自定义角色实现最小权限。权限将资源(数据库/集合)与允许的操作(find、insert、update 等)配对。授予所需的最低权限——报表用户只需要特定集合的 'find'。自定义角色存储在 admin 数据库中,可跨用户重用。

mongodb

# built-in roles:
#   read, readWrite, dbAdmin, dbOwner, userAdmin  (per database)
#   readAnyDatabase, readWriteAnyDatabase,         (all databases)
#   userAdminAnyDatabase, dbAdminAnyDatabase
#   root (superuser), clusterAdmin, backup, restore

# create a custom role with fine-grained privileges
db.createRole({
  role: "analyst",
  privileges: [
    { resource: { db: "mydb", collection: "reports" },
      actions: ["find"] },
    { resource: { db: "mydb", collection: "" },
      actions: ["listCollections"] }
  ],
  roles: []
})

# grant a role to a user
db.grantRolesToUser("appUser", [{ role: "analyst", db: "mydb" }])
db.revokeRolesFromUser("appUser", [{ role: "analyst", db: "mydb" }])

TLS/SSL 加密

TLS 加密传输中的数据——生产环境必需。requireTLS 拒绝非 TLS 连接。对于内部 PKI,使用 x.509 证书认证(无密码——证书即凭证)。$external 数据库存储 x.509 用户。在生产中始终使用真实 CA(或内部 CA);自签名证书仅用于测试。

mongodb

# enable TLS in mongod.conf
net:
  tls:
    mode: requireTLS
    certificateKeyFile: /etc/mongodb/server.pem
    CAFile: /etc/mongodb/ca.pem

# connect with TLS and verify the server cert
mongosh "mongodb://host:27017/mydb" \
  --tls --tlsCAFile /etc/mongodb/ca.pem

# connect with a client certificate (x.509 auth)
mongosh "mongodb://host:27017/mydb" \
  --tls --tlsCertificateKeyFile client.pem \
  --tlsCAFile ca.pem \
  --authenticationMechanism MONGODB-X509 \
  --authenticationDatabase '$external'

# generate a self-signed cert for testing
openssl req -x509 -newkey rsa:4096 -nodes \
  -keyout server.key -out server.crt -days 365

网络与审计

bindIp 限制 MongoDB 监听的网络接口——永远不要将 MongoDB 直接暴露到互联网(bindIp: 0.0.0.0 + 无认证已导致无数次数据泄露)。审计(仅企业版)记录与安全相关的操作以符合合规要求。社区版可以使用操作系统级工具(auditd)或应用级日志作为替代。

mongodb

# restrict network interface in mongod.conf
net:
  bindIp: 127.0.0.1,10.0.0.5    # never 0.0.0.0 in production
  port: 27017

# enable auditing (enterprise feature)
auditLog:
  destination: file
  format: JSON
  path: /var/log/mongodb/audit.log
  filter: '{ atype: { $in: ["authenticate","createCollection","dropDatabase"] } }'

# view current authentication mechanisms
db.runCommand({ getCmdLineOpts: 1 })

# enable FIPS mode (government compliance)
setParameter:
  fipsMode: true

字段级加密 (CSFLE)

CSFLE 在客户端加密敏感字段,使服务器、备份和日志永远不会包含明文——最强的数据保护。确定性加密支持等值查询;随机加密更安全但不可查询。自动加密(通过驱动程序模式)对应用代码透明。加密密钥在 KMS(AWS KMS、本地密钥等)中管理。

mongodb

# Client-Side Field Level Encryption encrypts fields
# before sending to the server — server never sees plaintext

# define a data key in the key vault
const clientEncryption = db.getMongo().getClientEncryption()
const keyId = clientEncryption.createDataKey(
  "local", { keyAltNames: ["mainKey"] }
)

# encrypt a field explicitly
const ssn = clientEncryption.encrypt(
  keyId, "123-45-6789", "deterministic"
)

# store encrypted, query with encrypted value
db.patients.insertOne({ name: "Alice", ssn: ssn })
db.patients.find({
  ssn: clientEncryption.encrypt(keyId, "123-45-6789", "deterministic")
})

# automatic CSFLE uses a schema to encrypt per a JSON schema
# { encryptMetadata: { keyId: [keyId], algorithm: "AEAD_AES_256_CBC..." } }

备份与恢复

mongodump 与 mongorestore

mongodump 创建逻辑备份(BSON 文件)——可移植但对大数据集很慢,且没有 --oplog 时跨集合不是时间点一致的。--drop 在恢复前删除现有集合。--archive + --gzip 生成单个压缩文件。对于大型生产备份,优先使用文件系统快照或 Percona Backup 而非 mongodump。

mongodb

# dump an entire database
mongodump --db=mydb --out=/backup/$(date +%F)

# dump a single collection
mongodump --db=mydb --collection=users --out=/backup

# dump with a query filter
mongodump --db=mydb --collection=logs \
  --query='{"date":{"$gte":{"$date":"2025-01-01T00:00:00Z"}}}'

# compress the output archive
mongodump --archive=/backup/full.gz --gzip

# restore from a dump
mongorestore --db=mydb --drop /backup/2025-01-01/mydb
mongorestore --archive=/backup/full.gz --gzip

mongoexport 与 mongoimport

mongoexport/mongoimport 处理 JSON/CSV——用于与其他系统的数据交换,不用于备份(它们会丢失 ObjectId 和 Date 精度等 BSON 类型)。备份请使用 mongodump。CSV 导出适合电子表格和 BI 工具。始终指定 --fields 以获得可预测的 CSV 列顺序。默认情况下 JSON 导出限制为每行一个文档(--jsonArray 用于单个数组)。

mongodb

# export to JSON (human-readable)
mongoexport --db=mydb --collection=users \
  --out=users.json

# export to CSV
mongoexport --db=mydb --collection=users \
  --type=csv --fields=name,email,age --out=users.csv

# export with a query filter
mongoexport --db=mydb --collection=users \
  --query='{"status":"active"}' --out=active.json

# import from JSON
mongoimport --db=mydb --collection=users --file=users.json

# import from CSV with header row
mongoimport --db=mydb --collection=users \
  --type=csv --headerline --file=users.csv

文件系统快照

文件系统快照是大型数据集最快的备份方法——它们一致地捕获整个数据目录。fsyncLock 简短地阻塞写入并刷新到磁盘,使快照一致。无论数据大小如何,快照几乎是即时的。恢复就像把文件复制回去一样简单——比 mongorestore 快得多。最适合 LVM、ZFS 或云块存储(EBS 快照)。

mongodb

# snapshot the dbPath volume for a consistent backup
# 1. flush writes to disk and lock (brief)
db.fsyncLock()

# 2. take a filesystem snapshot (LVM, ZFS, EBS, etc.)
lvcreate --snapshot --size 1G --name mongosnap /dev/vg/data

# 3. unlock the database
db.fsyncUnlock()

# 4. mount the snapshot and copy or back it up
mount /dev/vg/mongosnap /mnt/snap
rsync -a /mnt/snap/ /backup/

# restore: stop mongod, replace dbPath, start mongod
# (fastest restore method for large datasets)

时间点恢复 (oplog)

时间点恢复 (PITR) 在特定时刻重建数据库状态——在意外删除或错误迁移后至关重要。使用 --oplog 进行定期基础备份,然后将 oplog 重放到所需时间戳(--oplogLimit 在错误操作之前停止)。PITR 需要副本集(oplog 仅存在于副本集成员上)。

mongodb

# 1. take a base backup (snapshot or mongodump --oplog)
mongodump --archive=/backup/base.gz --oplog --gzip

# 2. note the time of the accident
# 3. replay the oplog up to just before the error

# dump the oplog from a secondary
mongodump --db=local --collection=oplog.rs \
  --query='{"ts":{"$gt":{"$timestamp":{"t":1700000000,"i":1}}}}' \
  --out=/oplog_dump

# restore the base backup, then replay oplog
mongorestore --archive=/backup/base.gz --oplogReplay \
  --oplogLimit=1700000123:1   # stop before this timestamp

# for a running replica set, use mongorestore --oplogReplay
# with a BSON oplog file

Atlas 备份与云选项

MongoDB Atlas 提供带时间点恢复的托管连续备份——对云部署最简单。Percona Backup for MongoDB (PBM) 是自建集群领先的开源备份工具,支持跨副本集和分片集群的物理备份和 PITR。始终测试恢复程序——无法恢复的备份不是备份。

mongodb

# MongoDB Atlas: continuous backups with PITR
# configured via UI or Atlas CLI:
atlas backups policies create [clusterId] ...

# on-demand snapshot
atlas backups snapshots create [clusterId] \
  --description "pre-migration" \
  --retentionDays 7

# restore an Atlas snapshot
atlas backups restores start [clusterId] \
  --snapshotId [snapId] \
  --deliveryType automated \
  --targetClusterName [target]

# Percona Backup for MongoDB (PBM) — open source
pbm backup --type=physical
pbm restore [snapshotName]
pbm pitr restore --time "2025-01-15T14:30:00"

性能与优化

分析慢查询

数据库分析器将慢于 slowms(默认 100ms)的操作记录到 system.profile(固定大小集合)。级别 2 记录所有操作——适用于调试但影响性能。查看 millis(持续时间)、nreturned(返回文档数)和 docsExamined 来查找低效查询。完成后禁用分析器。

mongodb

# enable the database profiler
# levels: 0=off, 1=slow ops only, 2=all ops
db.setProfilingLevel(1, { slowms: 100 })

# view profiled operations
db.system.profile.find().sort({ ts: -1 }).limit(5)

# find the slowest operations
db.system.profile.find({ millis: { $gt: 1000 } })
  .sort({ millis: -1 })

# profile a specific collection
db.system.profile.find({ ns: "mydb.users" })

# get current profiling level
db.getProfilingStatus()

# clear the profile collection
db.system.profile.drop()

explain() 深入

explain() 是你的主要调优工具。COLLSCAN 表示全集合扫描——添加索引。docsExamined 与 nReturned 的比率是关键效率指标:如果检查 10000 个文档只返回 10 个,索引就很差。allPlansExecution 显示规划器为何选择其计划。索引查询的相等、排序和范围字段(ESR 规则)。

mongodb

# three verbosity modes
db.users.find({age:30}).explain("queryPlanner")     # plan only
db.users.find({age:30}).explain("executionStats")    # + timing
db.users.find({age:30}).explain("allPlansExecution") # + rejected plans

# key metrics to check:
#   winningPlan.stage: COLLSCAN (bad) vs IXSCAN (good)
#   totalKeysExamined: index entries scanned
#   totalDocsExamined: documents fetched
#   nReturned: documents returned
#   executionTimeMillis: total time
#   works: number of work units

# ideal ratio: docsExamined ≈ nReturned
# if docsExamined >> nReturned -> missing or bad index

工作集与内存

性能取决于工作集(频繁访问的数据)能放入 RAM。如果 wiredTiger.cache 'pages read into cache' 很高,说明你在访问磁盘——增加 RAM 或添加索引以减少扫描。mongotop 显示哪些集合是 I/O 热点;mongostat 显示查询/插入/更新速率和页面错误。compact 回收删除/更新文档浪费的磁盘空间。

mongodb

# check the working set size
db.serverStatus().wiredTiger.cache
# look at: "maximum bytes configured", "current bytes"
#          "pages requested from disk" vs "pages read into cache"

# check storage stats
db.stats()
db.users.stats()
db.users.totalSize()   # data + index size in bytes

# mongotop: time spent reading/writing per collection
mongotop 30   # every 30 seconds

# mongostat: live server stats
mongostat --rowcount 10  # 10 iterations

# compact a collection (reclaims disk, blocks writes pre-4.4)
db.runCommand({ compact: "users" })

连接池

每个 MongoDB 连接消耗服务器内存——数千个连接会降低性能。使用连接池(驱动程序维护可重用池),maxPoolSize 根据你的负载调整(通常 50-100)。在整个应用中共享单个 MongoClient——永远不要为每个请求创建一个。设置 maxIdleTimeMS 以回收负载均衡器后面的陈旧连接。

mongodb

# each connection consumes memory (~1MB) on the server
# limit connections in the driver's connection string
mongodb://host:27017/mydb?maxPoolSize=50

# Node.js driver example
const { MongoClient } = require("mongodb")
const client = new MongoClient(uri, {
  maxPoolSize: 50,        // max connections
  minPoolSize: 5,         // keep-alive connections
  maxIdleTimeMS: 30000,   // close idle conns after 30s
  serverSelectionTimeoutMS: 5000
})

# view current connections on the server
db.serverStatus().connections
# { current: 120, available: 880, totalCreated: 200 }

批量写入与批处理大小

批量写入将操作分组为单次往返——比单独操作快得多。unordered:true 最大化吞吐量(出错后继续,在分片集群上并行化)。对于大规模导入,以 1000-5000 个文档为一组进行批处理,以平衡内存和往返次数。带 ordered:false 的 insertMany 是加载数据的最快方式。

mongodb

# ordered bulk write (stops on first error)
db.users.bulkWrite([
  { insertOne: { document: { name: "A" } } },
  { updateOne: { filter: {n:"B"}, update: {$set:{age:1}} } },
  { deleteOne: { filter: { n: "C" } } }
], { ordered: true })

# unordered bulk write (continues on errors, faster)
db.users.bulkWrite(ops, { ordered: false })

# insertMany is a bulk insert — batch large loads
const docs = Array.from({length: 100000}, (_,i) => ({ i }))
const batch = 1000
for (let i = 0; i < docs.length; i += batch) {
  db.big.insertMany(docs.slice(i, i + batch), { ordered: false })
}

变更流

监听变更

变更流通过跟踪 oplog 让应用实时响应数据变化。每个事件包含操作类型、完整文档(对于带 fullDocument 选项的插入/更新)和恢复令牌。存储恢复令牌以便在重启后恢复,不会遗漏或重复事件。需要副本集或分片集群。

mongodb

# open a change stream on a collection
const stream = db.users.watch()

# process change events (Node.js driver)
stream.on("change", (event) => {
  console.log(event.operationType)  // insert|update|delete|replace
  console.log(event.fullDocument)   // the new document
  console.log(event.documentKey)    // the _id
  console.log(event.updateDescription)  // updated fields
})

# resume token enables restart from last position
const token = stream.resumeToken
// store token, then resume later:
const stream2 = db.users.watch([], { resumeAfter: token })

# close the stream
stream.close()

过滤变更事件

用聚合管道过滤变更流——只有匹配的事件到达你的应用,减少噪音。默认情况下,更新只包含更改的字段,不包含完整文档;fullDocument:'updateLookup' 获取当前文档(每个事件额外查询)。fullDocumentBeforeChange (6.0+) 包含变更前的文档——适用于审计日志。

mongodb

# filter with an aggregation pipeline
const stream = db.users.watch([
  { $match: { operationType: "update" } },
  { $match: { "updateDescription.updatedFields.status": "active" } },
  { $project: { fullDocument: 1, documentKey: 1 } }
])

# watch specific operations
{ $match: { operationType: { $in: ["insert", "update"] } } }

# full document lookup on update (fetches current doc)
db.users.watch([], {
  fullDocument: "updateLookup"   // include latest doc on updates
})

# fullDocumentBeforeChange (MongoDB 6.0+)
db.users.watch([], {
  fullDocumentBeforeChange: "whenAvailable"
})

数据库/集群变更流

变更流可以监听单个集合、整个数据库或整个集群。数据库和集群流在每个事件中包含命名空间 (ns),让你知道变更发生在哪里。使用 startAtOperationTime 从 oplog 中的特定点重放变更——适用于在停机后回填或恢复遗漏的事件。

mongodb

# watch all collections in a database
const stream = db.watch()

# watch all databases in a cluster
const stream = client.watch()

# events include the ns (namespace) field
stream.on("change", (e) => {
  console.log(e.ns.db + "." + e.ns.coll)
  console.log(e.operationType)
})

# useful for audit logging, cache invalidation,
# cross-system sync, and real-time analytics

# start at a specific timestamp
db.users.watch([], {
  startAtOperationTime: Timestamp(1700000000, 1)
})

恢复令牌与可靠性

恢复令牌保证至少一次交付——在处理每个事件后持久化它们,以便重启时准确地从你离开的地方恢复。resumeAfter 和 startAfter 类似;startAfter 即使在集合失效事件(如删除)后也能工作。始终处理错误并使用最后一个令牌重新连接。令牌是不透明的——将其视为黑盒。

mongodb

# each event has a resume token (_id)
stream.on("change", (e) => {
  saveToStore(e._id)        // persist the token
  processEvent(e)
})

# after a restart, resume from the last token
const lastToken = loadFromStore()
const stream = db.users.watch([], {
  resumeAfter: lastToken     // resume after this event
})

# alternative: start after a cluster time
db.users.watch([], { startAfter: lastToken })

# handle errors and reconnect
stream.on("error", (err) => {
  console.error("stream error", err)
  // reconnect with the last saved token
})

# the token is opaque — store it, don't parse it

变更流用例

变更流支持实时架构:缓存失效(删除陈旧缓存条目)、实时通知(WebSocket 推送)、审计日志记录(为合规记录每次变更)和 ETL/同步(复制到 Elasticsearch、数据仓库或其他系统)。它们用干净的响应式事件驱动模型取代了手动轮询和触发器。

mongodb

# 1. Cache invalidation: clear cache when data changes
stream.on("change", (e) => {
  redis.del("user:" + e.documentKey._id)
})

# 2. Real-time notifications: push to WebSocket clients
stream.on("change", (e) => {
  io.emit("user-updated", e.fullDocument)
})

# 3. Audit log: record all changes to a separate store
stream.on("change", (e) => {
  auditLog.insertOne({
    op: e.operationType,
    doc: e.fullDocument,
    at: new Date()
  })
})

# 4. ETL: sync to a search index or warehouse
stream.on("change", async (e) => {
  await elasticsearch.index(e.fullDocument)
})

管理与工具

服务器状态与监控

serverStatus() 是主要监控来源——跟踪连接、操作计数器、缓存命中率和复制延迟。currentOp 显示运行中的操作;killOp 通过 opid 终止长时间运行的查询。设置监控代理(Prometheus、Datadog、Atlas)持续收集这些指标。关键警报:连接数接近上限、缓存命中率下降、复制延迟激增。

mongodb

# overall server status
db.serverStatus()

# key metrics to monitor:
db.serverStatus().connections       # active/idle connections
db.serverStatus().opcounters         # queries/inserts/updates/sec
db.serverStatus().wiredTiger.cache   # cache hit ratio (aim >95%)
db.serverStatus().metrics.document   # docs inserted/returned/deleted
db.serverStatus().repl               # replication state

# current operations (find long-running ops)
db.currentOp({ secs_running: { $gte: 5 } })

# kill a long-running operation
db.killOp(opid)

# host info (CPU, OS, memory)
db.hostInfo()

数据库与集合统计

stats() 显示存储详情:dataSize(逻辑大小)、storageSize(物理大小,经 WiredTiger 压缩后——通常小 50-80%)和 indexSize。estimatedDocumentCount 即时(来自集合元数据)但是近似值;countDocuments({}) 精确但会扫描。仪表板使用估算计数,计费使用精确计数。监控存储增长趋势以规划容量。

mongodb

# database-level stats
db.stats()
db.stats(1024*1024)   # in MB

# collection-level stats
db.users.stats()
db.users.dataSize()       # uncompressed data size
db.users.storageSize()    # on-disk size (with compression)
db.users.totalIndexSize() # total index size
db.users.totalSize()      # data + indexes

# count documents (accurate, may be slow)
db.users.countDocuments({})

# estimated count (fast, from metadata)
db.users.estimatedDocumentCount()

# list collections with sizes
db.runCommand({ listCollections: 1 })

压缩与修复

WiredTiger 重用释放的磁盘空间,因此很少需要 compact——仅在删除集合大部分数据后需要。compact 阻塞集合上的操作(4.4 之前)或在线运行(4.4+)。repairDatabase 是破坏性的最后手段,会重写所有数据——始终先备份。对于副本集上的空间回收,重新同步从节点(删除其数据并让其重新复制)。

mongodb

# compact: reclaim disk space from deleted/updated docs
# (WiredTiger reuses freed space, so rarely needed)
db.runCommand({ compact: "users" })

# fullDatabase compact (compact every collection)
db.getCollectionNames().forEach(c => {
  db.runCommand({ compact: c })
})

# repairDatabase (legacy, rewrites all data — last resort)
# Stops the server! Back up first.
mongod --repair --dbpath /data/db

# check data integrity (validates BSON)
db.users.validate({ full: true })

# reclaim space by re-syncing a secondary instead
# (safer than repair on a running primary)

mongotop 与 mongostat

mongostat 提供服务器负载的实时概览(插入、查询、缓存使用、连接)——类似于 MongoDB 的 top/vmstat。mongotop 显示哪些集合消耗最多 I/O 时间——对于查找热点集合非常宝贵。两者都随 MongoDB 捆绑提供。对于生产监控,将 mongod 日志和 serverStatus 指标发送到监控系统进行历史分析。

mongodb

# mongostat: live server stats (like vmstat for MongoDB)
mongostat
mongostat --host=host:27017 --rowcount 10
# columns: insert/query/update/delete/getmore/command
#          dirty (cache %), used (cache %), conn, time

# mongotop: time spent per collection (read/write)
mongotop
mongotop 30          # every 30 seconds
mongotop --host=host:27017

# columns: ns (namespace), total/read/write time (ms)
# helps identify which collections are I/O hot

# log slow queries in the mongod log
# set in mongod.conf:
# operationProfiling:
#   mode: slowOp
#   slowOpThresholdMs: 100

配置与启动

生产 MongoDB 通过配置文件运行(mongod.conf,使用 YAML)。关键设置:cacheSizeGB 限制 WiredTiger 的 RAM(默认使用 50% RAM——在共享服务器上显式设置),bindIp 限制网络访问,authorization 启用认证,replSetName/clusterRole 启用复制/分片。始终审查并版本控制你的 mongod.conf。serverCmdLineOpts 显示活动配置。

mongodb

# start mongod with a config file
mongod --config /etc/mongod.conf

# typical mongod.conf (YAML)
storage:
  dbPath: /var/lib/mongodb
  wiredTiger:
    engineConfig:
      cacheSizeGB: 4        # limit RAM usage
  journal:
    enabled: true
net:
  port: 27017
  bindIp: 127.0.0.1
security:
  authorization: enabled
replication:
  replSetName: rs0
sharding:
  clusterRole: shardsvr

# view current configuration
db.serverCmdLineOpts()
db.runCommand({ getCmdLineOpts: 1 })

驱动程序与集成

Node.js 驱动

Node.js 驱动基于 async/await,完全支持 ES6+。始终使用连接池(每个应用一个 MongoClient,重用)。游标是异步可迭代的。使用 for-await-of 迭代而不将所有文档加载到内存。在关闭时关闭客户端。驱动支持事务、变更流和所有聚合功能。

mongodb

const { MongoClient } = require("mongodb")

const client = new MongoClient(
  "mongodb://localhost:27017",
  { maxPoolSize: 50 }
)

async function main() {
  await client.connect()
  const db = client.db("mydb")
  const users = db.collection("users")

  // insert
  await users.insertOne({ name: "Alice", age: 30 })

  // find with a cursor
  const cursor = users.find({ age: { $gte: 18 } })
  for await (const doc of cursor) {
    console.log(doc.name)
  }

  // update
  await users.updateOne(
    { name: "Alice" },
    { $inc: { age: 1 } }
  )
}

main().finally(() => client.close())

Python 驱动 (PyMongo)

PyMongo 是官方 Python 驱动。它是同步的;对于异步使用 Motor(asyncio)或 Beanie(ODM)。PyMongo 字典是普通 Python 字典——不需要特殊对象。聚合管道是字典列表。上下文管理器事务模式确保自动提交/中止。始终使用连接池(默认 MongoClient 会池化连接)。

mongodb

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017")
db = client["mydb"]
users = db["users"]

# insert
users.insert_one({"name": "Alice", "age": 30})

# find
for doc in users.find({"age": {"$gte": 18}}).sort("age", -1):
    print(doc["name"])

# update
users.update_one(
    {"name": "Alice"},
    {"$inc": {"age": 1}}
)

# aggregation
pipeline = [
    {"$group": {"_id": "$city", "count": {"$sum": 1}}},
    {"$sort": {"count": -1}}
]
for doc in users.aggregate(pipeline):
    print(doc)

# use a context manager for transactions
with client.start_session() as session:
    with session.start_transaction():
        users.update_one({"n":"A"}, {"$inc":{"bal":-100}}, session=session)
        users.update_one({"n":"B"}, {"$inc":{"bal":100}}, session=session)

从驱动创建索引

从应用的设置/迁移脚本创建索引,而不是在每个请求上。createIndexes(复数)在一次调用中创建多个索引。在生产中,大集合上的索引构建可能很慢——MongoDB 4.2+ 默认非阻塞构建索引。将索引创建脚本与模式迁移一起版本控制。

mongodb

// Node.js: create indexes
await db.collection("users").createIndex({ email: 1 }, { unique: true })
await db.collection("users").createIndex({ name: "text" })

// Python
users.create_index([("email", 1)], unique=True)
users.create_index([("name", "text")])

// list indexes
const indexes = await db.collection("users").indexes()

// drop an index
await db.collection("users").dropIndex("email_1")

// create indexes in bulk
await db.collection("orders").createIndexes([
  { key: { customerId: 1 } },
  { key: { status: 1, date: -1 }, name: "status_date" }
])

连接字符串选项

连接字符串以查询参数形式携带所有客户端配置。mongodb+srv 使用 DNS SRV 记录发现主机——Atlas 使用此格式;它简化了连接字符串并在节点变更时自动更新。retryWrites=true(现代驱动默认)在网络抖动后自动重试幂等写入。设置合理的超时以避免连接挂起。

mongodb

# standard connection string
mongodb://user:pass@host1:27017,host2:27017/mydb?replicaSet=rs0

# common options (as query params):
#   replicaSet=rs0            - replica set name
#   readPreference=secondary  - route reads to secondaries
#   readConcernLevel=majority - read committed data
#   w=majority                - write concern
#   journal=true              - wait for journal flush
#   maxPoolSize=50            - max connections
#   connectTimeoutMS=5000     - connection timeout
#   socketTimeoutMS=30000     - socket timeout
#   authSource=admin          - db for credentials
#   tls=true                  - enable TLS
#   retryWrites=true          - retry on network errors

# SRV connection string (Atlas, uses DNS)
mongodb+srv://user:[email protected]/mydb

ODM 与高级库

ODM(对象文档映射器)添加模式、验证和更高级别的 API。Mongoose(Node.js)最受欢迎——它强制执行 MongoDB 原生缺乏的结构。Beanie 是领先的异步 Python ODM。ODM 非常适合快速开发和团队一致性,但会增加开销并可能隐藏 MongoDB 的强大功能。对于性能关键的应用,使用原始驱动。

mongodb

// Mongoose (Node.js ODM with schemas)
const mongoose = require("mongoose")
const userSchema = new mongoose.Schema({
  name: { type: String, required: true },
  age: { type: Number, min: 0 }
})
const User = mongoose.model("User", userSchema)
await User.create({ name: "Alice", age: 30 })
await User.find({ age: { $gte: 18 } })

# Beanie (Python async ODM)
from beanie import Document
class User(Document):
    name: str
    age: int
await User(name="Alice", age=30).insert()
await User.find(User.age >= 18).to_list()

# Mongose vs raw driver:
# ODMs add schemas, validation, and middleware
# but add overhead and a learning curve
# use the raw driver for max performance/control

GridFS 与大文件

GridFS 基础

GridFS 通过将文件拆分为块(默认 255KB)存储在 fs.chunks 中,元数据存储在 fs.files 中,从而存储超过 16MB BSON 文档限制的文件。它实际上是 MongoDB 之上的文件系统。用于超过 16MB 但仍受益于 MongoDB 复制和分片的文件。对于真正的大型媒体,考虑 S3 或 CDN。

mongodb

# GridFS stores files larger than 16MB (the BSON doc limit)
# by splitting them into chunks

# store a file via mongofiles (CLI)
mongofiles --db=mydb put large_video.mp4
mongofiles --db=mydb list
mongofiles --db=myfs get large_video.mp4

# two collections are used:
#   fs.files  - file metadata (one doc per file)
#   fs.chunks - file data in 255KB chunks

# query file metadata
db.fs.files.find().pretty()
db.fs.files.findOne({ filename: "large_video.mp4" })

# query chunks for a file
db.fs.chunks.find({ files_id: fileId }).sort({ n: 1 })

从驱动使用 GridFS

驱动程序将 GridFS 暴露为流式 API(GridFSBucket)——适合大文件,因为它们永远不会完全加载到内存中。上传/下载是流,你管道传输到/从文件或 HTTP 响应。上传上的 metadata 像任何文档字段一样可查询。GridFS 块在下载时自动重新组装。桶名(默认为 'fs')可配置。

mongodb

// Node.js: upload and download with streams
const { MongoClient, GridFSBucket } = require("mongodb")
const fs = require("fs")

const bucket = new GridFSBucket(db)

// upload a file
fs.createReadStream("video.mp4")
  .pipe(bucket.openUploadStream("video.mp4", {
    metadata: { uploadedBy: "alice" }
  }))

// download a file
bucket.openDownloadStreamByName("video.mp4")
  .pipe(fs.createWriteStream("downloaded.mp4"))

// delete a file
await bucket.delete(fileId)

// find files
const files = await bucket.find({}).toArray()

时间序列集合

时间序列集合(5.0+)针对顺序、时间排序的数据(IoT 传感器、指标、日志)进行了优化。它们在底层以类似列式的格式存储数据,实现 30-50% 的压缩和比常规集合更快的范围查询。granularity 应与你的数据间隔匹配。TTL(expireAfterSeconds)自动清除旧数据。使用 $dateTrunc 进行时间桶聚合。

mongodb

# create a time series collection (MongoDB 5.0+)
db.createCollection("temperatures", {
  timeseries: {
    timeField: "timestamp",
    metaField: "sensorId",
    granularity: "seconds"   // seconds|minutes|hours
  },
  expireAfterSeconds: 2592000   // auto-delete after 30 days
})

# insert measurements
db.temperatures.insertMany([
  { timestamp: ISODate(), sensorId: "s1", value: 22.5 },
  { timestamp: ISODate(), sensorId: "s1", value: 22.7 }
])

# query by time range
db.temperatures.find({
  sensorId: "s1",
  timestamp: { $gte: ISODate("2025-01-01") }
})

# aggregate averages per hour
db.temperatures.aggregate([
  { $group: {
    _id: { $dateTrunc: { date: "$timestamp", unit: "hour" } },
    avg: { $avg: "$value" }
  }}
])

固定集合

固定集合保持插入顺序,在满时覆盖最旧的文档——非常适合日志、最近活动流和环形缓冲区。它们支持高吞吐量插入和可尾随游标(类似 tail -f)。限制:你不能删除单个文档或将文档增长到超过其原始大小。变更流是实时尾随的现代替代方案。

mongodb

# capped collections have a fixed size and overwrite old docs
db.createCollection("logs", {
  capped: true,
  size: 5242880,    # 5MB max
  max: 10000        # optional: max document count
})

# inserts only (no arbitrary deletes/updates)
db.logs.insertOne({ msg: "started", t: new Date() })

# high-throughput tail with a tailable cursor
const cursor = db.logs.find().addOption(2)  // tailable
while (true) {
  if (cursor.hasNext()) printjson(cursor.next())
}

# convert a regular collection to capped
db.runCommand({
  convertToCapped: "events",
  size: 1000000
})

BSON 大小与文档限制

16MB BSON 文档限制是硬上限——使用 Object.bsonsize() 检查文档大小。大数组(数千个元素)使文档膨胀并减慢更新(MongoDB 在更新时重写整个文档)。对于无界数据(评论、评论、日志条目),使用带引用的单独集合。对于超过 16MB 的二进制大对象,使用 GridFS 或带 URL 引用的外部存储。

mongodb

# maximum BSON document size: 16MB
# check a document's size (in bytes)
Object.bsonsize(db.users.findOne())

# large arrays/fields should be referenced, not embedded
# a product with 10000 reviews -> use a separate collection

# check if a document is too large
const doc = { /* ... huge doc ... */ }
const size = Object.bsonsize(doc)
if (size > 16 * 1024 * 1024) {
  throw new Error("Document exceeds 16MB BSON limit")
}

# GridFS bypasses the limit by chunking
# alternatively, store large blobs externally (S3)
# and keep only a URL reference in the document

SQL 对比与迁移

SQL 到 MongoDB 映射

这个映射是 SQL 开发者学习 MongoDB 最快的方式。关键区别:MongoDB 没有固定模式(集合中的文档可以不同),'连接'被内嵌相关数据或使用 $lookup 取代(比 SQL 连接慢——设计模式以最小化它们)。SQL GROUP BY 映射到 $group 聚合阶段。

mongodb

# SQL -> MongoDB terminology
# database          -> database
# table             -> collection
# row               -> document
# column            -> field
# index             -> index
# join              -> $lookup (or embed)
# primary key       -> _id (ObjectId by default)
# foreign key       -> reference (ObjectId field)
# group by          -> $group
# select            -> find / $project
# where             -> $match

# SQL query -> MongoDB query
# SELECT * FROM users WHERE age >= 18
db.users.find({ age: { $gte: 18 } })

# SELECT name, age FROM users
db.users.find({}, { name: 1, age: 1, _id: 0 })

# SELECT * FROM users ORDER BY age DESC LIMIT 10
db.users.find().sort({ age: -1 }).limit(10)

# SELECT city, COUNT(*) FROM users GROUP BY city
db.users.aggregate([
  { $group: { _id: "$city", count: { $sum: 1 } } }
])

从 SQL 迁移

迁移不是 1:1 的表到集合映射——模式应该为 MongoDB 的文档模型重新设计。内嵌小的、独占的、频繁一起访问的数据(用户的个人资料+地址)。对大型、共享或无界数据(客户的订单)保留引用。MongoDB 的 Relational Migrator 工具自动化了大部分分析。规划模式,不要只是移植。

mongodb

# step 1: export SQL data to JSON/CSV
# (mysqldump, pg_dump, or a script)

# step 2: transform relational data to documents
# SQL:
#   users(id, name, dept_id)
#   departments(id, name)
# MongoDB (embed small, related data):
db.users.insertOne({
  _id: 1,
  name: "Alice",
  department: { id: 10, name: "Engineering" }  # embedded
})

# step 3: for large related data, keep references
db.orders.insertOne({
  _id: 101,
  customerId: 1,    # reference, not embedded
  items: [...]      # embed order items (bounded)
})

# tools: MongoDB Relational Migrator (GUI), mongoimport
# https://www.mongodb.com/.../relational-migrator

事务与 SQL ACID

MongoDB 提供 ACID:单文档操作始终是原子的(不需要事务),多文档事务(4.0+)提供跨集合的完整 ACID。关键区别:SQL 默认在每个语句上启用 ACID,而 MongoDB 鼓励将相关数据保持在一个文档中的模式设计(默认原子),仅在真正需要时使用事务。优先内嵌以避免事务开销。

mongodb

# SQL: ACID by default on every statement
BEGIN TRANSACTION;
UPDATE accounts SET balance = balance - 100 WHERE name = 'Alice';
UPDATE accounts SET balance = balance + 100 WHERE name = 'Bob';
COMMIT;

# MongoDB: single-document ops are atomic (ACID)
db.accounts.updateOne({name:"Alice"}, {$inc:{balance:-100}})
# (one document = one atomic operation)

# MongoDB: multi-document ACID (transactions, replica set)
const s = db.getMongo().startSession()
s.startTransaction()
db.accounts.updateOne({name:"Alice"},{$inc:{balance:-100}},{session:s})
db.accounts.updateOne({name:"Bob"},{$inc:{balance:100}},{session:s})
s.commitTransaction()

# prefer embedding to avoid transactions entirely
# (embedded data updates atomically in one document)

规范化与反规范化

SQL 规范化以避免重复;MongoDB 反规范化以提高读取速度。权衡:反规范化数据读取更快但更新更难(多个文档)。反规范化很少更改的数据(国家名称、产品目录),引用经常更改的数据(价格、库存)。正确的平衡取决于你的读写比率——监控并调整。

mongodb

# SQL: normalize to avoid duplication (3NF)
#   users(id, name, city_id)
#   cities(id, name, country_id)
#   countries(id, name)
# -> 4 joins to get a user's country

# MongoDB: denormalize for read performance
db.users.insertOne({
  name: "Alice",
  city: "NYC",
  country: "USA"        # denormalized — duplicate but fast reads
})

# trade-off: updates must touch multiple documents
# when a city's name changes:
db.users.updateMany(
  { "city": "New York" },
  { $set: { "city": "NYC" } }
)

# rule: duplicate data that changes rarely,
# reference data that changes often

何时使用 MongoDB 与 SQL

没有 universally'最好'的数据库。MongoDB 擅长灵活模式、层次数据和水平扩展。SQL 擅长关系完整性、复杂连接和繁重的事务工作负载。许多现代应用是多语言持久化——MongoDB 用于目录/内容/用户生成数据,PostgreSQL 用于金融交易。根据工作负载匹配数据库,而不是跟风。

mongodb

# CHOOSE MongoDB when:
# - schema is flexible or evolving rapidly
# - data is naturally hierarchical (embed)
# - you need horizontal scaling (sharding)
# - reads >> writes and you can denormalize
# - document-oriented domain (CMS, product catalogs, IoT)

# CHOOSE SQL (PostgreSQL/MySQL) when:
# - schema is fixed and relational (banking, ERP)
# - you need complex multi-table joins frequently
# - transactions span many tables constantly
# - strong consistency is non-negotiable
# - your team knows SQL well

# HYBRID: use both — MongoDB for product catalog/comments,
# SQL for orders/payments. Many large apps are polyglot.

# AVOID MongoDB for:
# - heavy multi-entity transactional workloads
# - apps requiring complex cross-table joins as core feature

高级聚合

$merge 与 $out(物化视图)

$out 用管道输出替换目标集合(破坏性)。$merge 更灵活——它可以插入、替换或与现有文档组合,实现增量物化视图。$merge 更适合计划刷新:只更新新/更改的数据。物化视图预计算昂贵的聚合以实现快速仪表板查询。

mongodb

# $out: write aggregation results to a NEW collection
db.orders.aggregate([
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $out: "customer_totals" }
])

# $merge: merge results into an EXISTING collection (flexible)
db.orders.aggregate([
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $merge: {
    into: "customer_totals",
    on: "_id",
    whenMatched: "replace",    // replace|keepExisting|merge|fail
    whenNotMatched: "insert"   // insert|discard|fail
  }}
])

# build a materialized view refreshed on a schedule
# (cron job or change-stream triggered)
db.daily_stats.aggregate([
  { $match: { date: today } },
  { $group: { _id: "$category", count: { $sum: 1 } } },
  { $merge: { into: "daily_stats", whenMatched: "replace" } }
])

窗口函数 ($setWindowFields)

$setWindowFields (5.0+) 将 SQL 窗口函数引入 MongoDB:运行总计、排名、移动平均和 lag/lead。partitionBy 是 PARTITION BY,sortBy 是 ORDER BY,window 定义帧。范围窗口使用值(例如最近 7 天),而文档窗口使用行计数。对时间序列分析和报表至关重要。

mongodb

# $setWindowFields: window functions (MongoDB 5.0+)
# like SQL OVER() — compute values across a window of docs
db.sales.aggregate([{
  $setWindowFields: {
    partitionBy: "$region",       # like PARTITION BY
    sortBy: { date: 1 },          # like ORDER BY
    output: {
      runningTotal: { $sum: "$amount", window: { documents: ["unbounded", "current"] } },
      rank: { $rank: {} },
      denseRank: { $denseRank: {} },
      movingAvg: { $avg: "$amount", window: { range: [-7, "current"], unit: "day" } },
      lagValue: { $shift: { by: -1, output: "$amount" } }
    }
  }
}])

# window: documents (count-based) or range (value-based)
# ["unbounded", "current"] = from start to current row
# [-3, "current"] = last 3 docs including current

$map、$filter 与 $reduce

$map、$filter 和 $reduce 是用于文档内转换的函数式数组操作符。$map 转换每个元素,$filter 保留匹配元素,$reduce 将数组聚合为一个值。'as' 变量(默认 $$this)命名当前元素。这些避免了许多数组操作的 $unwind——更快更干净。

mongodb

# $map: transform each array element
{ $project: {
  squares: { $map: {
    input: "$nums",
    as: "n",
    in: { $multiply: ["$$n", "$$n"] }
  }}
}}

# $filter: keep elements matching a condition
{ $project: {
  adults: { $filter: {
    input: "$users",
    as: "u",
    cond: { $gte: ["$$u.age", 18] }
  }}
}}

# $reduce: fold an array into a single value
{ $project: {
  total: { $reduce: {
    input: "$items",
    initialValue: 0,
    in: { $add: ["$$value", "$$this.price"] }
  }}
}}

# variables: $$value (accumulator), $$this (current element), $$n (named)

$facet 与多管道

$facet 在单个查询中对同一输入并行运行多个子管道——非常适合需要数据多种视图(细分、统计、Top N)的仪表板。每个子管道独立。结果是一个文档,每个子管道一个字段。这取代了多次往返并确保一致的快照。

mongodb

# run several aggregations on the same data in one pass
db.products.aggregate([{
  $facet: {
    "byCategory": [
      { $group: { _id: "$category", count: { $sum: 1 } } },
      { $sort: { count: -1 } }
    ],
    "priceStats": [
      { $group: { _id: null, avg: { $avg: "$price" }, max: { $max: "$price" } } }
    ],
    "topRated": [
      { $sort: { rating: -1 } },
      { $limit: 5 },
      { $project: { name: 1, rating: 1, _id: 0 } }
    ],
    "totalCount": [{ $count: "n" }]
  }
}])
# result: a single doc with all four arrays/objects

$bucket 与 $bucketAuto

$bucket 将文档分类到显式值范围(类似 SQL CASE WHEN)——非常适合价格层级、年龄组和直方图。$bucketAuto 让 MongoDB 计算边界以产生目标数量的桶(适合在不知道分布时的数据探索)。边界是左包含、右排除的。

mongodb

# $bucket: group documents into explicit ranges
db.products.aggregate([{
  $bucket: {
    groupBy: "$price",
    boundaries: [0, 50, 100, 500, 1000],
    default: "other",
    output: {
      count: { $sum: 1 },
      names: { $push: "$name" },
      avgPrice: { $avg: "$price" }
    }
  }
}])
# buckets: [0,50), [50,100), [100,500), [500,1000), other

# $bucketAuto: let MongoDB choose the boundaries
db.products.aggregate([{
  $bucketAuto: {
    groupBy: "$price",
    buckets: 5,           # target 5 buckets
    output: { count: { $sum: 1 }, avg: { $avg: "$price" } }
  }
}])
# good for histograms and data exploration

监控与故障排除

需要监控的关键指标

MongoDB 的四个黄金信号:连接数(接近上限表示泄漏或缺少池化)、缓存命中率(从磁盘读取的页面表示 RAM 压力)、复制延迟(滞后的从节点有故障转移数据丢失风险)和队列深度(锁竞争)。在所有四个上设置警报。缓存命中率下降是头号性能警告信号。

mongodb

# connections (watch for connection leaks)
db.serverStatus().connections
# { current, available, totalCreated }

# operation rates (queries/inserts/updates/deletes per sec)
db.serverStatus().opcounters

# WiredTiger cache (disk reads = cache misses)
db.serverStatus().wiredTiger.cache
# "bytes currently in the cache" vs "maximum bytes configured"
# "pages read into cache" should be low (cache hits)

# replication lag (critical for HA)
rs.printSecondaryReplicationInfo()

# queue depth (operations waiting for locks)
db.serverStatus().globalLock.currentQueue

# open cursors (watch for leaks)
db.serverStatus().metrics.cursor

解决锁竞争

MongoDB 使用文档级锁定(WiredTiger),因此竞争很少见——但未索引的查询、巨大的排序和长时间的事务仍会阻塞。currentOp 找到罪魁祸首;killOp 停止它。根本原因通常是缺少索引(运行 explain)。大集合上的模式迁移可能长时间锁定——在低流量时段或小批量执行。

mongodb

# check current queue (operations waiting)
db.serverStatus().globalLock.currentQueue
# { readers, writers, total }

# find long-running operations
db.currentOp({
  secs_running: { $gte: 5 },
  op: { $ne: "none" }
})

# kill a long-running op blocking others
db.killOp(opid)

# common causes of lock contention:
# - unindexed queries (COLLSCAN holds locks)
# - large in-memory sorts
# - long transactions
# - schema migrations on large collections

# check for slow ops in the log
db.adminCommand({ getLog: "global" }).log
  .filter(l => l.includes(" Slow query"))

内存与磁盘分析

WiredTiger 的缓存是内存前线——如果它充满了脏页,写入会变慢(刷新到磁盘)。如果 'pages read into cache' 很高,增加 RAM 或添加索引以减少扫描。删除未使用的索引——它们消耗磁盘和内存并减慢写入。监控存储增长以预测容量需求。在分片集群上,巨型块(太大无法迁移)表示分片键不佳。

mongodb

# check storage engine memory usage
db.serverStatus().wiredTiger.cache
# "maximum bytes configured" = cache limit (default 50% RAM)
# "tracked dirty bytes" = dirty pages awaiting flush
# target: dirty < 20%, cache usage < 95%

# disk space per collection
db.users.stats().storageSize
db.users.totalIndexSize()

# find the largest collections
db.getCollectionNames().map(c => ({
  name: c, size: db.getCollection(c).storageSize()
})).sort((a,b) => b.size - a.size).slice(0, 5)

# check for jumbo chunks (sharded clusters)
sh.status(true)

# review index sizes (drop unused indexes)
db.users.getIndexes().map(i => ({
  name: i.name, size: db.users.stats().indexSizes[i.name]
}))

日志分析

mongod 日志记录慢操作、选举、错误等。设置 slowOpThresholdMs 以捕获超过 SLO 的查询。getLog 检索内存中的环形缓冲区;磁盘上的日志在重启后持久存在。logRotate 归档当前日志(与 logrotate 一起用于每日轮换)。临时增加组件详细程度以进行深度调试(例如选举问题用 replication: 2),然后重置。

mongodb

# view recent log entries
db.adminCommand({ getLog: "global" })

# log slow queries (set threshold)
db.adminCommand({
  setParameter: 1,
  logComponentVerbosity: { query: { verbosity: 1 } }
})

# mongod.conf: log slow operations
# operationProfiling:
#   mode: slowOp
#   slowOpThresholdMs: 100
#   slowOpSampleRate: 1.0

# grep the mongod log for slow queries
grep "Slow query" /var/log/mongodb/mongod.log
grep "COMMAND" /var/log/mongodb/mongod.log | tail -50

# rotate logs
db.adminCommand({ logRotate: 1 })

# set log verbosity for a component
db.adminCommand({
  setParameter: 1,
  logComponentVerbosity: { replication: 2 }
})

常见错误与解决方案

这些是最常见的 MongoDB 错误。重复键错误来自 _id 或唯一索引冲突——在应用代码中处理。聚合内存限制用 allowDiskUse 修复(更慢但可用)。事务中的 WriteConflict 在竞争下是正常的——重试。游标超时意味着你的批处理太慢——使用更小的批次或 noCursorTimeout(但始终关闭游标)。

mongodb

# ERROR: "E11000 duplicate key"
# cause: _id or unique index conflict
# fix: check the duplicate value, use upsert, or catch the error

# ERROR: "QueryExceededMemoryLimitNoDiskUseAllowed"
# cause: aggregation stage exceeded 100MB RAM
# fix: add { allowDiskUse: true } or add $match early
db.big.aggregate([...], { allowDiskUse: true })

# ERROR: "BSONObjTooLarge"
# cause: document > 16MB
# fix: use GridFS or split the document

# ERROR: "WriteConflict" (transaction)
# cause: two transactions modified the same doc
# fix: retry the transaction (TransientTransactionError)

# ERROR: "ReplicaSetNoPrimary"
# cause: no primary (election in progress)
# fix: wait for election; check rs.status()

# ERROR: "cursor not found"
# cause: cursor timed out (10 min default)
# fix: use noCursorTimeout or process faster

MongoDB 速查表

入门

连接到 MongoDB

数据库基础

集合

BSON 数据类型

Shell 辅助命令与帮助

CRUD 操作

插入文档

查询文档

更新文档

更新操作符

删除文档

数组更新操作符

查询文档

比较操作符

逻辑操作符

元素与求值操作符

数组查询操作符

游标方法与排序

索引

创建索引

查看与管理索引

特殊索引类型

执行计划与查询计划

索引最佳实践

聚合框架

聚合管道基础

$match、$project 与 $limit

$group 与累加器

$lookup(连接)

$unwind、$sort 与 $facet

全文搜索与地理空间

全文搜索

地理空间 2dsphere

地理空间距离与操作符

正则表达式与模式匹配

通配符与灵活查询

事务

ACID 事务

事务读写关注

重试逻辑与错误

原子单文档操作

两阶段提交(模式)

数据建模

内嵌与引用

一对多关系

多对多关系

模式设计模式

模式验证

副本集

副本集基础

读偏好

写关注

仲裁者与选举

Oplog 与复制内部机制

分片

分片基础

分片键选择

区域与标签分片

块管理

分片集群组件

安全与认证

认证与用户

内置角色与自定义角色

TLS/SSL 加密

网络与审计

字段级加密 (CSFLE)

备份与恢复

mongodump 与 mongorestore

mongoexport 与 mongoimport

文件系统快照

时间点恢复 (oplog)

Atlas 备份与云选项

性能与优化

分析慢查询

explain() 深入

工作集与内存

连接池

批量写入与批处理大小