Go mgo 库多文档 Upsert 性能优化策略

package main

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

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

// 定义一个文档结构体
type Document struct {
    ID    bson.ObjectId `bson:"_id,omitempty"` // MongoDB 自动生成的 ID
    Key   string        `bson:"key"`           // 业务唯一键
    Value string        `bson:"value"`
    Count int           `bson:"count"`
}

func main() {
    // 1. 连接 MongoDB
    // 替换为你的 MongoDB 连接字符串
    session, err := mgo.Dial("mongodb://localhost:27017")
    if err != nil {
        log.Fatalf("Failed to connect to MongoDB: %v", err)
    }
    // 主会话在程序结束时关闭
    defer session.Close()

    // 设置会话模式，例如 ReadPreference
    session.SetMode(mgo.Primary, true)

    // 获取集合实例
    collection := session.DB("testdb").C("testcollection")

    // 2. 准备要 Upsert 的数据
    dataToUpsert := []Document{
        {Key: "item1", Value: "initialValueA", Count: 1},
        {Key: "item2", Value: "initialValueB", Count: 2},
        {Key: "item3", Value: "initialValueC", Count: 3},
        {Key: "item1", Value: "updatedValueA", Count: 10}, // 这将更新 item1
        {Key: "item4", Value: "initialValueD", Count: 4},
        {Key: "item2", Value: "updatedValueB", Count: 20}, // 这将更新 item2
    }

    var wg sync.WaitGroup
    // 使用带缓冲的通道收集所有 goroutine 可能产生的错误
    errChan := make(chan error, len(dataToUpsert))

    log.Printf("Starting concurrent upserts for %d documents...", len(dataToUpsert))
    start := time.Now()

    // 3. 使用 Goroutines 并发执行 Upsert
    for _, doc := range dataToUpsert {
        wg.Add(1)
        // 每次并发操作都克隆一个会话，确保并发安全
        // mgo.Session 不是并发安全的，每个 goroutine 必须使用其自身的会话副本
        go func(d Document, s *mgo.Session) {
            defer wg.Done()
            defer s.Close() // 确保克隆的会话在使用完毕后关闭

            // 定义查询条件，通常基于业务唯一键
            selector := bson.M{"key": d.Key}
            // 定义更新操作。如果文档不存在，mgo会插入一个包含selector和$set内容的文档。
            // 如果文档存在，则根据$set操作更新指定字段。
            update := bson.M{"$set": bson.M{"value": d.Value, "count": d.Count}}

            changeInfo, err := s.DB("testdb").C("testcollection").Upsert(selector, update)
            if err != nil {
                errChan <- fmt.Errorf("failed to upsert document with key '%s': %v", d.Key, err)
                return
            }
            // 根据 changeInfo 判断是插入还是更新
            if changeInfo.UpsertedId != nil {
                log.Printf("Inserted new document with key '%s', ID: %v", d.Key, changeInfo.UpsertedId)
            } else if changeInfo.Updated > 0 {
                log.Printf("Updated existing document with key '%s'", d.Key)
            } else {
                log.Printf("Upsert operation for key '%s' completed, but no change detected (might be identical data)", d.Key)
            }

        }(doc, session.Copy()) // 传递文档数据和克隆的会话
    }

    // 4. 等待所有 Goroutines 完成
    wg.Wait()
    close(errChan) // 关闭错误通道，以便后续遍历

    // 5. 检查并打印所有错误
    hasErrors := false
    for err := range errChan {
        log.Printf("Error during concurrent upsert: %v", err)
        hasErrors = true
    }

    duration := time.Since(start)
    if hasErrors {
        log.Printf("Concurrent upsert completed with errors in %v", duration)
    } else {
        log.Printf("All concurrent upserts completed successfully in %v", duration)
    }

    // 可选：验证数据
    log.Println("\n--- Verifying data in MongoDB ---")
    count, err := collection.Count()
    if err != nil {
        log.Printf("Failed to count documents: %v", err)
    } else {
        log.Printf("Total documents in collection: %d", count)
    }
    var results []Document
    err = collection.Find(nil).All(&results)
    if err != nil {
        log.Printf("Failed to retrieve documents: %v", err)
    } else {
        log.Printf("Documents in collection:")
        for _, doc := range results {
            log.Printf("  ID: %v, Key: %s, Value: %s, Count: %d", doc.ID, doc.Key, doc.Value, doc.Count)
        }
    }
}