MongoDB 的聚合管道（Aggregation Pipeline）是处理数据最强大的工具。你可以把它想象成一个数据处理工厂的流水线：文档（数据）从一端进入，经过一系列的处理阶段（Stage），最后从另一端输出你想要的结果。
在 Go 语言中，使用官方驱动 mongo-driver 操作聚合管道非常直观。下面我为你详细拆解它的核心用法。
🛠️ 核心概念：流水线 (Pipeline)
聚合管道由多个阶段（Stages）组成，每个阶段执行特定的操作，上一个阶段的输出是下一个阶段的输入。
在 Go 中，我们通常使用 bson.D（保持顺序的文档）或 bson.A（数组）来构建管道。
基本语法结构：
pipeline := mongo.Pipeline{{{"$阶段名", bson.D{{"参数", "值"}}}},// ... 更多阶段
}
cursor, err := collection.Aggregate(context.TODO(), pipeline)
🧩 常用阶段 (Stages)
这是你在开发中最常遇到的几个“工人”：
| 阶段 | 作用 | SQL 类比 |
|---|---|---|
$match |
过滤：筛选出符合条件的文档。 | WHERE |
$group |
分组：按某字段分组并进行统计（求和、计数等）。 | GROUP BY |
$project |
重塑：选择要显示的字段，或计算新字段。 | SELECT |
$sort |
排序：按字段升序或降序排列。 | ORDER BY |
$limit |
限制：只取前 N 条数据。 | LIMIT |
$unwind |
展开：将数组字段拆分成多条文档。 | (无直接对应) |
$lookup |
连表：关联另一个集合的数据（类似左连接）。 | LEFT JOIN |
💻 Go 语言实战示例
假设我们有一个名为 orders 的集合，存储了订单数据。
1. 准备数据
{ "_id": 1, "cust_id": "A", "amount": 100, "status": "A" }
{ "_id": 2, "cust_id": "B", "amount": 200, "status": "A" }
{ "_id": 3, "cust_id": "A", "amount": 50, "status": "B" }
2. 场景一：基础统计
需求：找出所有状态为 "A" 的订单，按客户 cust_id 分组，计算每个客户的总金额，并按金额降序排列。
import ("context""fmt""log""go.mongodb.org/mongo-driver/bson""go.mongodb.org/mongo-driver/mongo"
)func aggregateExample(collection *mongo.Collection) {// 构建管道pipeline := mongo.Pipeline{// 1. $match: 过滤状态为 "A" 的订单{{"$match", bson.D{{"status", "A"}}}},// 2. $group: 按 cust_id 分组，计算总金额// _id: "$cust_id" 表示分组依据是 cust_id 字段// totalAmount: { "$sum": "$amount" } 表示累加 amount 字段{{"$group", bson.D{{"_id", "$cust_id"}, {"totalAmount", bson.D{{"$sum", "$amount"}}},}}},// 3. $sort: 按总金额降序排列 (-1 表示降序){{"$sort", bson.D{{"totalAmount", -1}}}},}// 执行聚合cursor, err := collection.Aggregate(context.TODO(), pipeline)if err != nil {log.Fatal(err)}defer cursor.Close(context.TODO())// 遍历结果var results []bson.Mif err = cursor.All(context.TODO(), &results); err != nil {log.Fatal(err)}fmt.Println("统计结果：", results)// 输出示例: [{_id:A totalAmount:100} {_id:B totalAmount:200}]
}
3. 场景二：多表关联 ($lookup)
需求：查询订单，并把订单中的 cust_id 替换为 users 集合中对应的用户详细信息。
pipeline := mongo.Pipeline{{{"$lookup", bson.D{{"from", "users"}, // 要关联的集合名{"localField", "cust_id"}, // 当前集合的字段{"foreignField", "id"}, // 关联集合的字段{"as", "userInfo"}, // 关联结果存入的新字段名}}},
}cursor, err := collection.Aggregate(context.TODO(), pipeline)
// ... 处理 cursor ...
4. 场景三：字段重塑 ($project)
需求：只返回 cust_id 和 amount，并且把 amount 改名为 price，不显示 _id。
pipeline := mongo.Pipeline{{{"$project", bson.D{{"_id", 0}, // 0 表示排除该字段{"cust_id", 1}, // 1 表示包含该字段{"price": "$amount"}, // 将 amount 的值赋给新字段 price}}},
}
💡 性能优化小贴士
-
尽早过滤 (
$match前置)：
尽量把$match阶段放在管道的最前面。这样可以尽早减少数据量，让后续的阶段处理更少的数据，极大地提升效率。- 好：先
$match筛选出 100 条数据 -> 再$group。 - 坏：先
$group统计 100万条数据 -> 再$match。
- 好：先
-
利用索引：
$match和$sort阶段是可以利用数据库索引的。确保你的查询字段上有合适的索引。 -
使用
AllowDiskUse：
如果聚合操作处理的数据量非常大（超过 MongoDB 默认的 100MB 内存限制），需要在执行时开启磁盘使用权限：opts := options.Aggregate().SetAllowDiskUse(true) cursor, err := collection.Aggregate(ctx, pipeline, opts)
聚合管道是 MongoDB 的精髓，掌握它可以让你的 Go 后端代码变得非常简洁，把复杂的计算逻辑交给数据库去处理。