优化Go语言mgo库中MongoDB并发Upsert操作

package main

import (
    "fmt"
    "log"
    "sync"
    "time"

    "gopkg.in/mgo.v2"
    "gopkg.in/mgo.v2/bson"
)

// Document 结构体定义了我们将要操作的MongoDB文档模型
type Document struct {
    ID        bson.ObjectId `bson:"_id,omitempty"`
    Key       string        `bson:"key"`
    Value     string        `bson:"value"`
    Count     int           `bson:"count"`
    CreatedAt time.Time     `bson:"createdAt,omitempty"`
    UpdatedAt time.Time     `bson:"updatedAt,omitempty"`
}

func main() {
    // 1. 连接到MongoDB
    // 请根据您的MongoDB配置修改连接字符串
    session, err := mgo.Dial("mongodb://localhost:27017")
    if err != nil {
        log.Fatalf("连接MongoDB失败: %v", err)
    }
    defer session.Close() // 确保会话在程序结束时关闭

    // 设置会话模式，例如：Monotonic确保读取操作在同一连接上保持一致
    session.SetMode(mgo.Monotonic, true)

    // 获取数据库和集合
    collection := session.DB("testdb").C("mycollection")

    // 清空集合以便重复运行示例 (可选)
    // if _, err := collection.RemoveAll(nil); err != nil {
    //  log.Printf("清空集合失败: %v", err)
    // }

    // 2. 准备需要进行Upsert操作的文档数据
    docsToUpsert := []Document{
        {Key: "productA", Value: "Initial Product A", Count: 10},
        {Key: "productB", Value: "Initial Product B", Count: 20},
        {Key: "productC", Value: "Initial Product C", Count: 30},
        {Key: "productA", Value: "Updated Product A", Count: 12}, // 更新现有文档
        {Key: "productD", Value: "New Product D", Count: 40},
        {Key: "productB", Value: "Another Update B", Count: 25}, // 再次更新现有文档
    }

    // 3. 使用 sync.WaitGroup 管理并发协程
    var wg sync.WaitGroup
    start := time.Now() // 记录开始时间

    fmt.Println("开始并发执行Upsert操作...")

    for i, doc := range docsToUpsert {
        wg.Add(1) // 为每个协程增加计数器

        // 启动一个协程来处理每个文档的Upsert
        go func(index int, d Document) {
            defer wg.Done() // 协程完成后减少计数器

            // 定义查询选择器，通常用于查找文档的唯一键
            selector := bson.M{"key": d.Key}

            // 定义更新操作。
            // $set 用于设置或更新字段值。
            // $inc 用于原子性地增加字段值。
            // $setOnInsert 用于仅在文档被插入时设置字段值。
            change := bson.M{
                "$set": bson.M{
                    "value":     d.Value,
                    "updatedAt": time.Now(), // 每次更新时设置更新时间
                },
                "$inc": bson.M{
                    "count": 1, // 每次操作都增加计数
                },
                "$setOnInsert": bson.M{
                    "createdAt": time.Now(), // 仅在插入时设置创建时间
                },
            }

            // 执行Upsert操作
            _, err := collection.Upsert(selector, change)
            if err != nil {
                log.Printf("协程 %d: Upsert文档 (key: '%s') 失败: %v", index, d.Key, err)
            } else {
                fmt.Printf("协程 %d: 成功Upsert文档 (key: '%s')\n", index, d.Key)
            }
        }(i, doc) // 将循环变量作为参数传递给协程，避免闭包陷阱
    }

    wg.Wait() // 等待所有协程完成
    elapsed := time.Since(start)
    fmt.Printf("所有并发Upsert操作完成，耗时: %s\n", elapsed)

    // 4. (可选) 验证文档是否已正确更新或插入
    fmt.Println("\n验证集合中的文档:")
    var results []Document
    err = collection.Find(nil).All(&results)
    if err != nil {
        log.Printf("获取文档失败: %v", err)
    } else {
        for _, res := range results {
            fmt.Printf("ID: %s, Key: %s, Value: %s, Count: %d, CreatedAt: %s, UpdatedAt: %s\n",
                res.ID.Hex(), res.Key, res.Value, res.Count,
                res.CreatedAt.Format("2006-01-02 15:04:05"), res.UpdatedAt.Format("2006-01-02 15:04:05"))
        }
    }
}