> ## Content Index
> Fetch the complete content index at: https://blog.vercanti.com/llms.txt
> Use this file to discover other available public pages before exploring further.

# MongoDB 完全指南
- URL: https://blog.vercanti.com/mongodb-wan-quan-zhi-nan/
- Published: 2026-08-28T14:35:36.000Z
- Updated: 2026-08-28T14:59:09.000Z
- Description: 相关文档：MySQL基础完全指南(/mysql-ji-chu-wan-quan-zhi-nan/) Redis完全指南(/redis-wan-quan-zhi-nan/) Scrapy分布式采集(/scrapy-fen-bu-shi-cai-ji-wan-quan-zhi-nan/) 聚合管道是 MongoDB 最强大的查询能力，通过一系列阶段（Stage）对数据进行变换： 嵌入 vs 引用：一对少用嵌入，一对多用引用：嵌入文档（subdocument）读取一次 IO 就能拿到所有数据，适合一对少（如用户地址、商品规格）；引用（_id 外键 + $loo
- Author: yellowdog
- Tags: 数据库

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

相关文档：[MySQL基础完全指南](https://blog.vercanti.com/mysql-ji-chu-wan-quan-zhi-nan/) [Redis完全指南](https://blog.vercanti.com/redis-wan-quan-zhi-nan/) [Scrapy分布式采集](https://blog.vercanti.com/scrapy-fen-bu-shi-cai-ji-wan-quan-zhi-nan/)

---

## 1\. 基础概念

### MongoDB vs 关系型数据库

| 概念  | MongoDB            | 关系型数据库      |
| --- | ------------------ | ----------- |
| 数据库 | Database           | Database    |
| 集合  | Collection         | Table（表）    |
| 文档  | Document（BSON）     | Row（行）      |
| 字段  | Field              | Column（列）   |
| 主键  | \_id（ObjectId）     | PRIMARY KEY |
| 关联  | $lookup（聚合） / 嵌套文档 | JOIN        |
| 模式  | 无模式（灵活）            | 严格模式        |

### 适用场景

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

### 安装（Docker）

```bash
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

### 插入

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

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

```

### 查询

```js
// 查询所有
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: "alice@example.com" })

```

### 更新

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

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

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

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

```

### 更新操作符

| 操作符          | 说明              | 示例                                |
| ------------ | --------------- | --------------------------------- |
| $set         | 设置字段值           | { $set: { name: "Bob" } }         |
| $unset       | 删除字段            | { $unset: { old\_field: "" } }    |
| $inc         | 数值自增/减          | { $inc: { count: 1 } }            |
| $mul         | 数值乘以            | { $mul: { price: 1.1 } }          |
| $rename      | 重命名字段           | { $rename: { "old": "new" } }     |
| $push        | 向数组追加元素         | { $push: { tags: "js" } }         |
| $pull        | 从数组移除元素         | { $pull: { tags: "php" } }        |
| $addToSet    | 向数组追加（不重复）      | { $addToSet: { tags: "python" } } |
| $pop         | 移除数组头/尾         | { $pop: { items: 1 } } 移除最后一个     |
| $setOnInsert | 仅在 upsert 插入时设置 | 用于初始化字段                           |

### 删除

```js
db.users.deleteOne({ email: "alice@example.com" })
db.users.deleteMany({ is_active: false })

// 返回被删除的文档
db.users.findOneAndDelete({ email: "alice@example.com" })

```

---

## 3\. 查询操作符

### 比较操作符

| 操作符        | 说明          |
| ---------- | ----------- |
| $eq        | 等于（等同于直接写值） |
| $ne        | 不等于         |
| $gt / $gte | 大于 / 大于等于   |
| $lt / $lte | 小于 / 小于等于   |
| $in        | 在列表中        |
| $nin       | 不在列表中       |

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

```

### 逻辑操作符

```js
// 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 }] })

```

### 元素操作符

```js
// 字段是否存在
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"] } })

```

### 正则查询

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

```

---

## 4\. 聚合管道（Aggregation）

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

### 常用阶段

| 阶段             | 说明                 |
| -------------- | ------------------ |
| $match         | 过滤文档（尽量放最前面，利用索引）  |
| $group         | 分组统计               |
| $project       | 字段投影/变换            |
| $sort          | 排序                 |
| $limit / $skip | 分页                 |
| $lookup        | 关联其他集合（类似 JOIN）    |
| $unwind        | 展开数组字段             |
| $addFields     | 添加新字段              |
| $count         | 统计总数               |
| $facet         | 多管道并行（一次查询返回多维度结果） |

### 分组统计示例

```js
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 多维度统计

```js
// 一次请求返回列表 + 总数 + 分类统计
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\. 索引

```js
// 单字段索引
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")

```

### 执行计划

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

```

---

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

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

```

```python
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 集成

```python
# 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`）适合一对多且子文档独立访问频繁的场景（如订单-商品）。

```js
// 嵌入（一对少，地址不超过 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")` 确认索引命中情况。

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

// 验证是否命中索引
db.users.find({ email: "a@b.com" }).explain("executionStats")
// 看 winningPlan 是否为 IXSCAN

```

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

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

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

```

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

```python
# 错误：全量加载
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"` 确保多数节点已写入。

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

```

---

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

### ObjectId 与字符串转换

```python
from bson import ObjectId

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

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

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

```

### 用 Pydantic 管理文档结构

```python
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`。

```python
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()`，语义更清晰。

```python
# 错误：替换整个文档
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": "b@b.com"})

```

---

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

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

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

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

```js
// 正确：$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基础完全指南](https://blog.vercanti.com/mysql-ji-chu-wan-quan-zhi-nan/)
- [Redis完全指南](https://blog.vercanti.com/redis-wan-quan-zhi-nan/)
- [PostgreSQL完全指南](https://blog.vercanti.com/postgresql-wan-quan-zhi-nan/)
- [Elasticsearch完全指南](https://blog.vercanti.com/elasticsearch-wan-quan-zhi-nan/)
- [Scrapy分布式采集](https://blog.vercanti.com/scrapy-fen-bu-shi-cai-ji-wan-quan-zhi-nan/)