MongoDB 完全指南

官方文档：https://www.mongodb.com/docs/manual/
适用版本：MongoDB 7.0 / Motor 3.x（2026-05-07 核实）

相关文档：MySQL基础完全指南 Redis完全指南 Scrapy分布式采集

1. 基础概念

MongoDB vs 关系型数据库

适用场景

• 字段不固定的动态数据（用户配置、商品属性）
• 嵌套/树形数据（评论线程、菜单）
• 爬虫原始数据存储
• 日志、时序数据（配合 Time Series 集合）
• 高写入量、水平扩展需求

安装（Docker）

docker run -d \
  --name mongodb \
  -e MONGO_INITDB_ROOT_USERNAME=admin \
  -e MONGO_INITDB_ROOT_PASSWORD=password \
  -p 27017:27017 \
  mongo:7

# 连接
mongosh "mongodb://admin:password@localhost:27017"

2. 基础 CRUD

插入

// 插入单个文档
db.users.insertOne({
  name: "Alice",
  email: "[email protected]",
  age: 25,
  tags: ["python", "mongodb"],
  address: { city: "北京", district: "海淀区" }
})

// 批量插入
db.users.insertMany([
  { name: "Bob", email: "[email protected]", age: 30 },
  { name: "Charlie", email: "[email protected]", age: 28 },
])

查询

// 查询所有
db.users.find()

// 条件查询
db.users.find({ age: { $gt: 20 } })
db.users.find({ name: "Alice", is_active: true })

// 嵌套字段查询（点表示法）
db.users.find({ "address.city": "北京" })

// 数组查询
db.users.find({ tags: "python" })           // 包含 python
db.users.find({ tags: { $all: ["python", "mongodb"] } })  // 包含所有

// 投影（只返回指定字段）
db.users.find({}, { name: 1, email: 1, _id: 0 })

// 排序、分页
db.users.find().sort({ age: -1 }).skip(20).limit(10)

// 查询单个
db.users.findOne({ email: "[email protected]" })

更新

// 更新单个（第一个匹配的）
db.users.updateOne(
  { email: "[email protected]" },           // 条件
  { $set: { age: 26, "address.city": "上海" } }  // 更新操作
)

// 批量更新
db.users.updateMany(
  { is_active: false },
  { $set: { deleted_at: new Date() } }
)

// 更新或插入（upsert）
db.users.updateOne(
  { email: "[email protected]" },
  { $setOnInsert: { created_at: new Date() }, $set: { name: "New User" } },
  { upsert: true }
)

// 返回更新后的文档
db.users.findOneAndUpdate(
  { email: "[email protected]" },
  { $inc: { login_count: 1 } },    // 原子自增
  { returnDocument: "after" }      // 返回更新后的文档
)

更新操作符

删除

db.users.deleteOne({ email: "[email protected]" })
db.users.deleteMany({ is_active: false })

// 返回被删除的文档
db.users.findOneAndDelete({ email: "[email protected]" })

3. 查询操作符

比较操作符

db.users.find({ age: { $gte: 18, $lte: 65 } })
db.users.find({ status: { $in: ["active", "pending"] } })

逻辑操作符

// AND（默认多条件即 AND）
db.users.find({ age: { $gt: 20 }, is_active: true })

// OR
db.users.find({ $or: [{ age: { $lt: 18 } }, { is_active: false }] })

// NOT
db.users.find({ age: { $not: { $gt: 60 } } })

// NOR
db.users.find({ $nor: [{ age: { $lt: 18 } }, { is_active: false }] })

元素操作符

// 字段是否存在
db.users.find({ phone: { $exists: true } })
db.users.find({ deleted_at: { $exists: false } })

// 类型匹配
db.users.find({ age: { $type: "int" } })
db.users.find({ price: { $type: ["int", "double"] } })

正则查询

// 不区分大小写的正则
db.users.find({ name: { $regex: /^alice/i } })
db.users.find({ name: { $regex: "^alice", $options: "i" } })

4. 聚合管道（Aggregation）

聚合管道是 MongoDB 最强大的查询能力，通过一系列阶段（Stage）对数据进行变换：

常用阶段

分组统计示例

db.orders.aggregate([
  // 1. 过滤：只统计已完成的订单
  { $match: { status: "completed", created_at: { $gte: new Date("2026-01-01") } } },

  // 2. 分组：按用户统计订单数和总金额
  { $group: {
    _id: "$user_id",
    order_count: { $sum: 1 },
    total_amount: { $sum: "$amount" },
    avg_amount: { $avg: "$amount" },
    max_amount: { $max: "$amount" },
  }},

  // 3. 关联用户信息
  { $lookup: {
    from: "users",
    localField: "_id",
    foreignField: "_id",
    as: "user",
  }},

  // 4. 展开 user 数组（lookup 结果是数组）
  { $unwind: "$user" },

  // 5. 投影：整理输出字段
  { $project: {
    _id: 0,
    user_name: "$user.name",
    order_count: 1,
    total_amount: 1,
    avg_amount: { $round: ["$avg_amount", 2] },
  }},

  // 6. 排序
  { $sort: { total_amount: -1 } },

  // 7. 分页
  { $limit: 10 },
])

$facet 多维度统计

// 一次请求返回列表 + 总数 + 分类统计
db.products.aggregate([
  { $match: { is_active: true } },
  { $facet: {
    // 分支 1：分页数据
    items: [
      { $sort: { created_at: -1 } },
      { $skip: 0 },
      { $limit: 20 },
    ],
    // 分支 2：总数
    total: [{ $count: "count" }],
    // 分支 3：按分类统计数量
    by_category: [
      { $group: { _id: "$category", count: { $sum: 1 } } },
    ],
  }},
])

5. 索引

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

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

// 复合索引
db.orders.createIndex({ user_id: 1, status: 1, created_at: -1 })

// 部分索引（只索引满足条件的文档）
db.users.createIndex(
  { email: 1 },
  { partialFilterExpression: { is_active: true } }
)

// 过期索引（TTL，自动删除过期文档）
db.sessions.createIndex({ created_at: 1 }, { expireAfterSeconds: 86400 })  // 24小时后删除

// 文本索引（全文搜索）
db.articles.createIndex({ title: "text", content: "text" })

// 查看索引
db.users.getIndexes()

// 删除索引
db.users.dropIndex("email_1")

执行计划

db.users.find({ email: "[email protected]" }).explain("executionStats")
// 关注：
// winningPlan.stage: IXSCAN（索引扫描，好）/ COLLSCAN（全表扫描，差）
// executionStats.totalDocsExamined: 扫描文档数，越小越好
// executionStats.totalKeysExamined: 扫描索引条目数

6. Python 操作（Motor 异步驱动）

pip install motor  # MongoDB 异步驱动（基于 pymongo）

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient
from bson import ObjectId
from datetime import datetime

# 连接
client = AsyncIOMotorClient("mongodb://admin:password@localhost:27017")
db = client.mydb
users = db.users

# 插入
async def create_user(data: dict) -> str:
    result = await users.insert_one({
        **data,
        "created_at": datetime.utcnow(),
        "is_active": True,
    })
    return str(result.inserted_id)

# 查询单个
async def get_user(user_id: str) -> dict | None:
    doc = await users.find_one({"_id": ObjectId(user_id)})
    if doc:
        doc["id"] = str(doc.pop("_id"))
    return doc

# 查询列表（分页）
async def list_users(page: int = 1, page_size: int = 20) -> list[dict]:
    cursor = users.find(
        {"is_active": True},
        {"password": 0}   # 排除密码字段
    ).sort("created_at", -1).skip((page - 1) * page_size).limit(page_size)

    result = []
    async for doc in cursor:
        doc["id"] = str(doc.pop("_id"))
        result.append(doc)
    return result

# 更新
async def update_user(user_id: str, data: dict) -> bool:
    result = await users.update_one(
        {"_id": ObjectId(user_id)},
        {"$set": {**data, "updated_at": datetime.utcnow()}}
    )
    return result.modified_count > 0

# 删除
async def delete_user(user_id: str) -> bool:
    result = await users.delete_one({"_id": ObjectId(user_id)})
    return result.deleted_count > 0

# 聚合
async def get_user_stats() -> list[dict]:
    pipeline = [
        {"$match": {"is_active": True}},
        {"$group": {
            "_id": "$city",
            "count": {"$sum": 1},
            "avg_age": {"$avg": "$age"},
        }},
        {"$sort": {"count": -1}},
    ]
    return await users.aggregate(pipeline).to_list(length=None)

与 FastAPI 集成

# src/database.py
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from contextlib import asynccontextmanager
from fastapi import FastAPI

client: AsyncIOMotorClient | None = None

def get_db() -> AsyncIOMotorDatabase:
    return client.mydb

@asynccontextmanager
async def lifespan(app: FastAPI):
    global client
    client = AsyncIOMotorClient("mongodb://admin:password@localhost:27017")
    yield
    client.close()

最佳实践

嵌入 vs 引用：一对少用嵌入，一对多用引用：嵌入文档（subdocument）读取一次 IO 就能拿到所有数据，适合一对少（如用户地址、商品规格）；引用（_id 外键 + $lookup）适合一对多且子文档独立访问频繁的场景（如订单-商品）。

// 嵌入（一对少，地址不超过 5 个）
{ _id: "user_1", name: "Alice", addresses: [{ city: "北京", zip: "100000" }] }

// 引用（一对多，订单独立查询）
// users: { _id: "user_1" }
// orders: { _id: "order_1", user_id: "user_1", total: 199.00 }

对高频查询字段建索引，对 _id 以外的唯一字段建唯一索引：MongoDB 只有 _id 有默认索引，其余字段全表扫描。通过 explain("executionStats") 确认索引命中情况。

// 创建索引
db.users.createIndex({ email: 1 }, { unique: true })
db.orders.createIndex({ user_id: 1, created_at: -1 })  // 复合索引

// 验证是否命中索引
db.users.find({ email: "[email protected]" }).explain("executionStats")
// 看 winningPlan 是否为 IXSCAN

更新必须用 $set 等更新运算符，不直接替换文档：不带更新运算符的 update 会用新文档完整替换旧文档，丢失其他字段。

// 错误：替换整个文档
db.users.updateOne({ _id: id }, { name: "Bob" })

// 正确：只更新指定字段
db.users.updateOne({ _id: id }, { $set: { name: "Bob" } })

大量文档查询用游标分批处理，不用 toArray() 一次全加载：百万文档一次 toArray() 会撑爆内存。用 .batchSize() 控制每次网络传输量，用 async for 流式处理。

# 错误：全量加载
all_docs = await collection.find().to_list(length=None)

# 正确：游标迭代
async for doc in collection.find({}).batch_size(500):
    await process(doc)

写入重要数据时设置合适的 writeConcern：默认 w:1 只等待 Primary 确认，主节点宕机时有少量数据丢失风险。关键数据用 w:"majority" 确保多数节点已写入。

await collection.insert_one(
    {"user_id": 1, "payment": 999.00},
    session=session,
)
# Motor 通过连接字符串设置默认 writeConcern
# mongodb://host/?w=majority&journal=true

7. 最佳实践（代码示例）

ObjectId 与字符串转换

from bson import ObjectId

# ObjectId 转字符串（返回给前端）
str(doc["_id"])

# 字符串转 ObjectId（查询时）
ObjectId("507f1f77bcf86cd799439011")

# 验证是否合法 ObjectId
ObjectId.is_valid("507f1f77bcf86cd799439011")  # True

用 Pydantic 管理文档结构

from pydantic import BaseModel, Field
from bson import ObjectId
from datetime import datetime

class PyObjectId(str):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v):
        if not ObjectId.is_valid(v):
            raise ValueError("Invalid ObjectId")
        return str(v)

class UserDocument(BaseModel):
    id: PyObjectId | None = Field(default=None, alias="_id")
    name: str
    email: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        populate_by_name = True
        json_encoders = {ObjectId: str}

常见陷阱

陷阱：_id 的 ObjectId 无法直接 JSON 序列化

现象： FastAPI 或 Flask 返回文档时报 TypeError: Object of type ObjectId is not JSON serializable。

原因： ObjectId 是 BSON 类型，不是 Python 内置类型，标准库 json.dumps 不知道如何序列化它。

解决： 转换时 str(doc["_id"])，或使用自定义 JSON 编码器，或用 Pydantic model 配合 json_encoders。

from bson import ObjectId
import json

class ObjectIdEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, ObjectId):
            return str(obj)
        return super().default(obj)

json.dumps(doc, cls=ObjectIdEncoder)

# FastAPI + Pydantic v2
from pydantic import BaseModel, field_serializer
class UserOut(BaseModel):
    id: str
    name: str

陷阱：update_one 不加 $set 导致文档被完整替换

现象： 只想修改 name 字段，操作后文档中其他所有字段都消失了。

原因： 不带 $set 的 update_one 把第二个参数当作新文档直接替换旧文档，而不是部分更新。

解决： 所有字段级更新使用 $set；如果确实想替换整个文档用 replace_one()，语义更清晰。

# 错误：替换整个文档
await users.update_one({"_id": id}, {"name": "Bob"})

# 正确：只更新 name 字段
await users.update_one({"_id": id}, {"$set": {"name": "Bob"}})

# 明确替换整个文档（语义清晰）
await users.replace_one({"_id": id}, {"name": "Bob", "email": "[email protected]"})

陷阱：缺少索引导致聚合管道全集合扫描

现象： 聚合查询在小数据集上很快，数据增长到百万后变得极慢，甚至超时。

原因： 聚合管道的 $match 阶段如果放在前面且字段有索引，MongoDB 会在扫描前过滤文档；否则对整个集合做全扫描再过滤。

解决： 将 $match 放在管道的最开始；为 $match 中用到的字段创建合适的索引；用 explain() 验证。

// 正确：$match 放最前，且 user_id 有索引
db.orders.aggregate([
    { $match: { user_id: ObjectId("..."), status: "paid" } },
    { $group: { _id: "$product_id", total: { $sum: "$amount" } } }
])

// 创建复合索引支持此查询
db.orders.createIndex({ user_id: 1, status: 1 })

参见

• MySQL基础完全指南
• Redis完全指南
• PostgreSQL完全指南
• Elasticsearch完全指南
• Scrapy分布式采集