Forrest
返回归档
💈技术分享

MongoDB 总结

更新于 2025/08/2414 分钟阅读
文章目录

MongoDB是什么

MongoDB 是一个基于 分布式文件存储 的开源 NoSQL 数据库系统，由 C++ 编写的。MongoDB 提供了 面向文档 的存储方式，操作起来比较简单和容易，支持“无模式”的数据建模，可以存储比较复杂的数据类型，是一款非常流行的 文档类型数据库 。

在高负载的情况下，MongoDB 天然支持水平扩展和高可用，可以很方便地添加更多的节点/实例，以保证服务性能和可用性。在许多场景下，MongoDB 可以用于代替传统的关系型数据库或键/值存储方式，皆在为 Web 应用提供可扩展的高可用高性能数据存储解决方案。

MongoDB的存储结构

  • 文档（Document）：MongoDB 中最基本的单元，由 BSON 键值对（key-value）组成，类似于关系型数据库中的行（Row）。
  • 集合（Collection）：一个集合可以包含多个文档，类似于关系型数据库中的表（Table）。
  • 数据库（Database）：一个数据库中可以包含多个集合，可以在 MongoDB 中创建多个数据库，类似于关系型数据库中的数据库（Database）。

SQL 与 MongoDB 常见术语对比

SQL MongoDB
表（Table） 集合（Collection）
行（Row） 文档（Document）
列（Col） 字段（Field）
主键（Primary Key） 对象 ID（Objectid）
索引（Index） 索引（Index）
嵌套表（Embedded Table） 嵌入式文档（Embedded Document）
数组（Array） 数组（Array）

MongoDB的存储引擎

查看引擎

db.serverStatus().storageEngine

MongoDB的存储引擎有很多种，WiredTiger 是目前最常用的（默认）存储引擎，提供了高性能、压缩和事务支持。

  • 支持文档级别的并发控制（Document-Level Concurrency），允许多个操作同时修改集合中的不同文档，提高了并发性能。
  • 支持压缩（Compression），可以减少磁盘空间占用，支持 Snappy 和 Zlib 压缩算法。
  • 支持高效的缓存管理，使用内存缓存（Cache）来加速数据访问。
  • 支持事务（Transactions），从 MongoDB 4.0 开始支持多文档事务。

简单使用

使用docker-compose 搭建环境

mkdir -p /Users/Zhuanz/docker-compose/mongodb && cd /Users/Zhuanz/docker-compose/mongodb
cat <<EOF> /Users/Zhuanz/docker-compose/mongodb/docker-compose.yaml
version: '3'
services:
  mongodb:
    restart: always
    image: registry.cn-shanghai.aliyuncs.com/wanfei/mongo:7.0.12
    container_name: mongodb
    privileged: true    
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: root
      TZ: Asia/Shanghai
    volumes:
      - /Users/Zhuanz/docker-compose/mongodb/data:/data/db
    ports:
      - 27017:27017
    command: mongod --auth
EOF
docker-compose -f /Users/Zhuanz/docker-compose/mongodb/docker-compose.yaml up -d

image

基础的增删改查

  • 创建集合：db.createCollection("collectionName")
  • 插入数据：db.collectionName.insertOne() 或 db.collectionName.insertMany()
  • 查询数据：db.collectionName.find()
  • 更新数据：db.collection.updateOne(filter, update, options)或db.collection.updateMany(filter, update, options)
    • filter：用于查找文档的查询条件。
    • update：指定更新操作的文档或更新操作符。
    • options：可选参数对象，如 upsert、arrayFilters 等
  • 删除数据：db.collectionName.deleteOne() 或 db.collectionName.deleteMany()
  • 删除集合：db.collectionName.drop()
  • sort表示排序，1标识升序，-1标识降序
// 创建Database ts1
use ts1;
//创建Collection users
db.createCollection("users");

db.users.insert({
	name: "alice",
	age: 15,
	email: "alice@example.com"
});

db.users.insert({
	name: "bob",
	age: 20,
	email: "bob@gami.com"
})

db.users.insert({
	name: "Tom",
	age: 30,
	email: "tom@gmail.com"
});

db.users.update(
	{ name: "bob" },
	{ $set: { age: 30 } }
);

db.users.update(
  { "name": "bob" },
  { $inc: { "age": 1 } }
);

db.users.find({age: { $gt: 25 }});

db.users.deleteMany(
	{age: { $gt: 10 }}
);

db.users.find();
db.movies.insertOne({
  "title": "星际穿越",
  "director": "克里斯托弗·诺兰",
  "year": 2014,
  "genre": "科幻",
  "rating": 9.3,
  "duration": 169
});
//使用push向数组中添加元素
db.movies.updateOne(
  { "title": "星际穿越" },
  { $push: { "actors": "马修·麦康纳" } }
);
db.movies.updateOne(
  { "title": "星际穿越" },
  { $push: { "actors": "约翰福音" } }
);
//使用pull从数组中移除元素
db.movies.updateOne(
  { "title": "星际穿越" },
  { $pull: { "actors": "马修·麦康纳" } }
);

MongoDB的聚合操作

由多个阶段组成，每个阶段在文档通过管道时转换文档。每个阶段接收前一个阶段的输出，进一步处理数据，并将其作为输入数据发送到下一个阶段。

image

常用阶段操作符：

操作符 简述
$match 匹配操作符，用于对文档集合进行筛选
$project 投射操作符，用于重构每一个文档的字段，可以提取字段，重命名字段，甚至可以对原有字段进行操作后新增字段
$sort 排序操作符，用于根据一个或多个字段对文档进行排序
$limit 限制操作符，用于限制返回文档的数量
$skip 跳过操作符，用于跳过指定数量的文档
$count 统计操作符，用于统计文档的数量
$group 分组操作符，用于对文档集合进行分组
$unwind 拆分操作符，用于将数组中的每一个值拆分为单独的文档
$lookup 连接操作符，用于连接同一个数据库中另一个集合，并获取指定的文档，类似于 populate

语法格式

db.orders.aggregate([
   # 第一阶段：$match阶段按status字段过滤文档，并将status等于"A"的文档传递到下一阶段。
    { $match: { status: "A" } },
  # 第二阶段：$group阶段按cust_id字段将文档分组，以计算每个cust_id唯一值的金额总和。
    { $group: { _id: "$cust_id", total: { $sum: "$amount" } } }
])

实例

group 分组查询

db.createCollection("orders");

// 插入多条订单数据
db.orders.insertMany([
    { user_id: 1, amount: 100, status: "completed", date: ISODate("2023-10-01") },
    { user_id: 2, amount: 200, status: "completed", date: ISODate("2023-10-02") },
    { user_id: 1, amount: 150, status: "pending", date: ISODate("2023-10-03") },
    { user_id: 3, amount: 300, status: "completed", date: ISODate("2023-10-04") },
    { user_id: 2, amount: 50, status: "pending", date: ISODate("2023-10-05") }
]);

db.orders.find();

db.orders.aggregate([
	{$match: {status: "completed"}},
	{ $group: {
		_id: "$user_id",
		totalAmount: { $sum: "$amount" }
	}}
])

bucket 同组数据汇聚

db.users.insertMany([
	{name:"alice",age: 10},
	{name:"bob",age: 18},
	{name:"tom",age: 15},
	{name:"golang",age: 21},
	{name:"java",age: 30},
	{name:"cpp",age: 38}
])

db.users.find();

db.users.aggregate([{
	$bucket:{
		groupBy: "$age",
		boundaries: [10,20,30,40],
		default: "Other",
		output: {"count":{$sum:1}}
	}
}])

unwind 将数组展开

db.students.insertMany([
	{name:"x",scores:[{语文:"100"},{数学:"80"},{英语:"120"}]},
	{name:"t",scores:[{语文:"100"},{数学:"80"},{英语:"120"}]},	
]);

db.students.find()

db.students.aggregate([
	{$unwind: "$scores"}
])

lookup实现关联查询

db.users.insertMany([
  { _id: 1, name: "alice", age: 10 },
  { _id: 2, name: "bob", age: 18 },
  { _id: 3, name: "tom", age: 15 }
]);

db.orders.insertMany([
  { _id: 1, userId: 1, product: "Laptop", quantity: 1 },
  { _id: 2, userId: 1, product: "Mouse", quantity: 2 },
  { _id: 3, userId: 2, product: "Keyboard", quantity: 1 }
]);

db.users.aggregate([
	{
		$lookup:{
			from: "orders",
			localField: "_id",
			foreignField: "userId",
			as : "userOrders"
		}
	}
])

也可以搭配$unwind将userOrders数组拆分

db.users.aggregate([
	{
		$lookup:{
			from:"orders",
			localField: "_id",
			foreignField: "userId",
			as : "userOrders"
		}
	},
	{
		$unwind:{
			path: "$userOrders"
		}
	}
])

MongoDB 架构

Master-Slave

Master-Slave 由主从角色构成：

Master ( 主 )

可读可写，当数据有修改的时候，会将 Oplog 同步到所有连接的Salve 上去

Slave ( 从 )

只读，所有的 Slave 从 Master 同步数据，从节点与从节点之间不感知

image

Replica Set（副本集模式）

  • 数据多副本，在故障的时候，可以使用完的副本恢复服务。注意：这里是故障自动恢复；

  • 读写分离，读的请求分流到副本上，减轻主（Primary）的读压力；

  • 节点直接互有心跳，可以感知集群的整体状态；

    image

一、拓扑感知过程 (Topology Discovery)

初始发现：

  1. 种子连接：驱动程序从连接字符串获取种子节点地址
  2. hello 命令：连接任意种子节点，发送 hello 命令获取集群信息
  3. 拓扑构建：根据返回的成员列表、主节点信息等，在内存中构建完整拓扑视图

持续监控：

  • 驱动程序定期向所有节点发送心跳检测
  • 实时感知节点状态变化、故障转移等事件
  • 动态更新内存中的拓扑视图，确保请求路由准确

二、读写流程

写操作流程：

  1. 应用程序发起写请求
  2. 驱动程序查询拓扑视图，定位当前 Primary 节点
  3. 写请求直接发送到 Primary 节点
  4. Primary 执行写操作并记录到 Oplog
  5. Secondary 节点异步从 Oplog 复制数据
  6. Primary 根据写策略返回确认

读操作流程（读写分离示例）：

  1. 应用程序发起读请求，指定读偏好为 secondary
  2. 驱动程序从健康的 Secondary 节点中选择一个（通常选择延迟最低的）
  3. 读请求直接发送到选中的 Secondary 节点
  4. Secondary 节点执行查询并返回结果
  5. 驱动程序将结果返回给应用程序

核心特点：

  • 写操作必须经过 Primary 节点，保证强一致性
  • 读操作可根据读偏好策略灵活选择节点
  • 从 Secondary 读取可分担主节点压力，但可能存在数据延迟
  • 整个过程由客户端驱动程序自主完成，无需外部中间件

Sharding

image

MongoDB 的 sharding 是一种水平扩展方法，允许跨多台机器分布数据。sharding 架构由以下几个组件组成：

  • mongos: 这是查询路由器，充当应用程序和分片集群之间的接口。客户端连接到 mongos，而不是直接连接到各个分片
  • Config Servers
    • config Servers 维护着集群元数据，包括 Chunk 到 Shard 的映射关系
    • 这些映射存储在 Config Server 的 config.chunks 集合中
    • 每个 Chunk 记录包含其范围(min和max值)和所属的 Shard ID
  • Shard: 每个 shard 是存储部分数据的独立 MongoDB 实例或 replica set。实际数据存储在这里
  • 分片键(Shard Key): 用于在分片间分配数据的字段。MongoDB 使用分片键决定数据属于哪个分片。

工作原理

  • 数据分区: 基于分片键，将集合拆分为块(chunks)。
  • 分布式查询: mongos 会将查询路由到适当的分片。
  • 负载平衡: MongoDB 自动在分片之间平衡数据，确保数据均匀分布。
  • 自动故障转移: 每个分片通常是一个 replica set，提供高可用性。

拓扑感知过程

mongos 的拓扑感知：

  1. 启动初始化：mongos 从配置中获取 Config Servers 地址
  2. 元数据同步：连接 Config Servers，获取完整的集群元数据
    • Shard 列表和状态
    • 数据库和集合的分片配置
    • Chunk 分布映射（config.chunks 集合）
  3. 缓存构建：在内存中构建路由表缓存
  4. 持续更新：定期从 Config Servers 刷新元数据，保持路由信息最新

应用程序的连接：

  • 应用程序只需连接 mongos，无需了解底层 shard 分布
  • 驱动程序将 mongos 视为普通 MongoDB 实例进行连接

写操作流程：

  1. 应用请求：应用程序向 mongos 发送写请求
  2. 路由计算：mongos 根据 shard key 值计算目标 chunk
  3. shard 定位：查询路由表确定 chunk 所在的 shard
  4. 请求转发：mongos 将写请求转发到目标 shard
  5. shard 执行：目标 shard（通常是 replica set 的 primary）执行写操作
  6. 结果返回：shard 将执行结果返回给 mongos，mongos 再返回给应用程序

读操作流程：

单 shard 查询：

  1. mongos 根据查询条件中的 shard key 确定目标 shard
  2. 将查询直接路由到对应 shard
  3. shard 执行查询并返回结果

多 shard 查询（scatter-gather）：

  1. mongos 识别查询需要访问多个 shard
  2. 并行向相关 shard 发送查询请求
  3. 收集各 shard 的结果
  4. 在 mongos 层面进行结果合并、排序等处理
  5. 返回最终结果给应用程序

Sharding 与 Replica Sets 对比

相似点

  • 都提供高可用性和故障转移功能
  • 都支持自动恢复
  • 都使用 MongoDB 的原生复制机制

区别

特性 Sharding Replica Sets
主要目的 水平扩展，提高性能和存储容量 数据冗余和高可用性
数据分布 数据分布在多个分片上 完整数据集复制到所有节点
扩展方式 水平扩展（更多机器） 主要是垂直扩展（更好的硬件）
架构复杂性 较复杂，需要维护更多组件 相对简单
适用场景 大数据量，高并发 中小规模数据，需要冗余
查询路由 通过 mongos 路由 直接连接到 primary 或 secondary
管理开销 较高 较低

MongoDB的事物

MongoDB 事务遵循 ACID 属性（原子性、一致性、隔离性和持久性），允许对多个文档的操作作为一个单元执行，要么全部成功，要么全部失败。

支持的操作

事务中支持以下操作：

  • CRUD 操作 (创建、读取、更新、删除)
  • 批量写入操作
  • 查找和修改操作

不支持的操作

事务中不支持：

  • 创建/删除集合
  • 创建/删除索引
  • 删除数据库
  • 聚合 $out 和 $merge 阶段

事务配置参数

读关注 (Read Concern)

定义事务中读操作返回的数据的一致性级别：

  • "local": 返回最新的数据，无需确认是否已被多数节点确认
  • "majority": 返回已被大多数节点确认的数据
  • "snapshot": 返回同一时间点的多文档快照（事务中推荐）

写关注 (Write Concern)

定义写操作的确认要求：

  • { w: 1 }: 主节点确认
  • { w: "majority" }: 大多数节点确认（事务中推荐）
  • { j: true }: 写入操作必须写入到磁盘日志

读偏好 (Read Preference)

定义从哪些节点读取数据：

  • "primary": 从主节点读取（事务中必须）

事务限制与最佳实践

限制

  1. 时间限制: 事务必须在 60 秒内完成（可配置但不推荐超过默认值）
  2. 大小限制: 单个事务中的操作不能超过 16MB
  3. 资源锁: 过长的事务可能会导致资源锁定
  4. 性能影响: 事务会对性能产生一定影响

最佳实践

  1. 保持事务简短: 尽量减少事务中的操作数量和持续时间
  2. 避免大事务: 不要在一个事务中处理太多文档
  3. 合理设置超时: 根据业务需求设置适当的超时时间
  4. 异常处理: 始终包含错误处理和回滚逻辑
  5. 会话复用: 在可能的情况下复用会话以提高性能
  6. 索引优化: 确保事务中使用的查询有适当的索引支持

MongoDB的索引

查看索引

db.collection.getIndexes()

创建索引

// 创建唯一索引
db.collection.createIndex( { field: 1 }, { unique: true } )

// 创建后台运行的索引
db.collection.createIndex( { field: 1 }, { background: true } )

// 创建稀疏索引
db.collection.createIndex( { field: 1 }, { sparse: true } )

//创建hash索引
 db.collection.createIndex( { field: "hashed" } )

删除索引

// 删除指定的索引
db.collection.dropIndex( "indexName" )

// 删除所有索引
db.collection.dropIndexes()

使用Golang SDK

code 
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"go.mongodb.org/mongo-driver/mongo/readconcern"
	"go.mongodb.org/mongo-driver/mongo/readpref"
	"go.mongodb.org/mongo-driver/mongo/writeconcern"
)

// Account 表示银行账户
type Account struct {
	AccountID string  `bson:"accountId"`
	Balance   float64 `bson:"balance"`
	Owner     string  `bson:"owner"`
}

// Transaction 表示交易记录
type Transaction struct {
	From      string    `bson:"from"`
	To        string    `bson:"to"`
	Amount    float64   `bson:"amount"`
	Timestamp time.Time `bson:"timestamp"`
}

func main() {
	// 设置连接字符串
	uri := "mongodb://root:root@localhost:27017"

	client, err := mongo.Connect(context.Background(), options.Client().ApplyURI(uri))
	if err != nil {
		log.Fatalf("无法连接到MongoDB: %v", err)
	}
	defer client.Disconnect(context.Background())

	err = client.Ping(context.Background(), readpref.Primary())
	if err != nil {
		log.Fatalf("无法ping MongoDB: %v", err)
	}
	fmt.Println("成功连接到MongoDB!")

	// 获取数据库和集合
	bankDB := client.Database("bank")
	accountsCollection := bankDB.Collection("accounts")

	// 准备测试数据
	prepareTestData(accountsCollection)

	// 执行转账事务
	transferAmount(client, 100.0, "A123", "B456")

	// 显示转账后的账户状态
	displayAccounts(accountsCollection)
}

func prepareTestData(accountsCollection *mongo.Collection) {
	// 删除现有数据
	accountsCollection.DeleteMany(context.Background(), bson.M{})

	// 创建测试账户
	accounts := []interface{}{
		Account{AccountID: "A123", Balance: 1000.0, Owner: "张三"},
		Account{AccountID: "B456", Balance: 500.0, Owner: "李四"},
	}

	// 插入测试账户
	_, err := accountsCollection.InsertMany(context.Background(), accounts)
	if err != nil {
		log.Fatalf("无法创建测试数据: %v", err)
	}
	fmt.Println("测试数据已准备好")
}

func transferAmount(client *mongo.Client, amount float64, fromAccount, toAccount string) {
	// 创建会话
	session, err := client.StartSession()
	if err != nil {
		log.Fatalf("无法创建会话: %v", err)
	}
	defer session.EndSession(context.Background())

	// 事务选项
	txnOpts := options.Transaction().
		SetReadConcern(readconcern.Snapshot()).
		SetWriteConcern(writeconcern.New(writeconcern.WMajority()))

	// 执行事务
	err = mongo.WithSession(context.Background(), session, func(sessionContext mongo.SessionContext) error {
		if err := session.StartTransaction(txnOpts); err != nil {
			return fmt.Errorf("无法开始事务: %w", err)
		}

		bankDB := client.Database("bank")
		accountsCollection := bankDB.Collection("accounts")
		transactionsCollection := bankDB.Collection("transactions")

		// 1. 检查并从账户A扣款
		fromResult := accountsCollection.FindOneAndUpdate(
			sessionContext,
			bson.M{"accountId": fromAccount, "balance": bson.M{"$gte": amount}},
			bson.M{"$inc": bson.M{"balance": -amount}},
		)

		var fromAcc Account
		if err := fromResult.Decode(&fromAcc); err != nil {
			// 回滚事务
			session.AbortTransaction(sessionContext)
			return fmt.Errorf("账户 %s 余额不足或不存在: %w", fromAccount, err)
		}

		// 2. 向账户B存款
		toResult, err := accountsCollection.UpdateOne(
			sessionContext,
			bson.M{"accountId": toAccount},
			bson.M{"$inc": bson.M{"balance": amount}},
		)

		if err != nil || toResult.ModifiedCount != 1 {
			// 回滚事务
			session.AbortTransaction(sessionContext)
			return fmt.Errorf("无法更新账户 %s: %w", toAccount, err)
		}

		// 3. 记录交易历史
		_, err = transactionsCollection.InsertOne(
			sessionContext,
			Transaction{
				From:      fromAccount,
				To:        toAccount,
				Amount:    amount,
				Timestamp: time.Now(),
			},
		)

		if err != nil {
			// 回滚事务
			session.AbortTransaction(sessionContext)
			return fmt.Errorf("无法记录交易: %w", err)
		}

		// 提交事务
		if err := session.CommitTransaction(sessionContext); err != nil {
			return fmt.Errorf("无法提交事务: %w", err)
		}

		return nil
	})

	if err != nil {
		log.Printf("转账失败: %v", err)
	} else {
		fmt.Printf("成功从账户 %s 转账 %.2f 到账户 %s\n", fromAccount, amount, toAccount)
	}
}

// 显示账户信息
func displayAccounts(accountsCollection *mongo.Collection) {
	cursor, err := accountsCollection.Find(context.Background(), bson.M{})
	if err != nil {
		log.Fatalf("查询账户失败: %v", err)
	}
	defer cursor.Close(context.Background())

	fmt.Println("\n当前账户状态:")
	fmt.Println("------------------")
	for cursor.Next(context.Background()) {
		var account Account
		if err := cursor.Decode(&account); err != nil {
			log.Fatalf("解码账户数据失败: %v", err)
		}
		fmt.Printf("账户: %s, 所有者: %s, 余额: %.2f\n", account.AccountID, account.Owner, account.Balance)
	}
	fmt.Println("------------------")
}

需要注意的点

查询时

处理 mongo.ErrNoDocuments 错误，表示没有找到文档

var result bson.M
err = collection.FindOne(ctx, bson.D{{"name", "alice"}}).Decode(&result)
if err != nil {
    if err == mongo.ErrNoDocuments {
        fmt.Println("No documents found")
    } else {
        log.Fatal(err)
    }
}
fmt.Println("Found document:", result)

更新时

使用$set 更新指定字段，避免覆盖整个文档

filter := bson.D{{"name", "alice"}}
update := bson.D{{"$set", bson.D{{"age", 30}}}}
result, err := collection.UpdateOne(ctx, filter, update)
if err != nil {
    log.Fatal(err)
}
fmt.Println("Matched count:", result.MatchedCount, "Modified count:", result.ModifiedCount)

Replica

Reference

bookmark