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)

这是你在开发中最常遇到的几个“工人”：

💻 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.M
    if 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
    }}},
}

💡 性能优化小贴士

1. 尽早过滤 ($match 前置)：
   尽量把 $match 阶段放在管道的最前面。这样可以尽早减少数据量，让后续的阶段处理更少的数据，极大地提升效率。

   • 好：先 $match 筛选出 100 条数据 -> 再 $group。
   • 坏：先 $group 统计 100万条数据 -> 再 $match。

2. 利用索引：
   $match 和 $sort 阶段是可以利用数据库索引的。确保你的查询字段上有合适的索引。

3. 使用 AllowDiskUse：
   如果聚合操作处理的数据量非常大（超过 MongoDB 默认的 100MB 内存限制），需要在执行时开启磁盘使用权限：

   opts := options.Aggregate().SetAllowDiskUse(true)
   cursor, err := collection.Aggregate(ctx, pipeline, opts)
   

聚合管道是 MongoDB 的精髓，掌握它可以让你的 Go 后端代码变得非常简洁，把复杂的计算逻辑交给数据库去处理。