Go并发编程中MongoDB会话管理与Goroutine生命周期

package main

import (
    "fmt"
    "labix.org/v2/mgo"
    "labix.org/v2/mgo/bson"
    "time" // 引入time包用于模拟耗时操作
)

type User struct {
    Id    string `bson:"_id"` // MongoDB的_id字段
    Email string
}

type Post struct {
    Id          string `bson:"_id"`
    UserId      string `bson:"user_id"` // 关联用户ID
    Description string
}

// handleUser 函数处理单个用户的帖子
func handleUser(db *mgo.Database, user *User) {
    fmt.Println("处理用户 - ID:", user.Id, " EMAIL:", user.Email)

    result := Post{}
    // 模拟耗时操作，确保goroutine有时间执行
    time.Sleep(50 * time.Millisecond) 

    iter := db.C("posts").Find(bson.M{"user_id": user.Id}).Iter()

    for iter.Next(&result) {
        fmt.Println("  帖子 - ID:", result.Id, " 描述:", result.Description)
    }
    if err := iter.Close(); err != nil {
        fmt.Println("迭代器关闭错误:", err)
    }
}

func main() {
    session, err := mgo.Dial("localhost:27017") // 确保MongoDB服务运行在27017端口

    if err != nil {
        panic(err)
    }
    // 初始设置，插入一些测试数据
    // defer session.Close() // 暂时注释掉，看问题如何发生

    db := session.DB("mydb")

    // 清理旧数据并插入新数据
    db.C("users").DropCollection()
    db.C("posts").DropCollection()

    db.C("users").Insert(&User{Id: "user1", Email: "user1@example.com"})
    db.C("users").Insert(&User{Id: "user2", Email: "user2@example.com"})
    db.C("posts").Insert(&Post{Id: "post1_1", UserId: "user1", Description: "User1's first post"})
    db.C("posts").Insert(&Post{Id: "post1_2", UserId: "user1", Description: "User1's second post"})
    db.C("posts").Insert(&Post{Id: "post2_1", UserId: "user2", Description: "User2's first post"})

    fmt.Println("开始处理用户...")

    result := User{}
    iter := db.C("users").Find(nil).Iter()
    for iter.Next(&result) {
        // 尝试并发调用 handleUser
        go handleUser(db, &result) // 问题发生在这里
    }
    if err := iter.Close(); err != nil {
        fmt.Println("主迭代器关闭错误:", err)
    }

    // 如果不加任何同步机制，main函数会立即返回，导致goroutine无法完成
    // time.Sleep(1 * time.Second) // 临时解决方案，不推荐
    // session.Close() // 应该在所有goroutine完成后关闭
    fmt.Println("主函数即将退出...")
}

package main

import (
    "fmt"
    "labix.org/v2/mgo"
    "labix.org/v2/mgo/bson"
    "sync" // 引入sync包
    "time"
)

type User struct {
    Id    string `bson:"_id"`
    Email string
}

type Post struct {
    Id          string `bson:"_id"`
    UserId      string `bson:"user_id"`
    Description string
}

// handleUser 函数现在接收一个独立的会话副本
func handleUser(session *mgo.Session, user *User, wg *sync.WaitGroup) {
    defer wg.Done() // goroutine完成时通知WaitGroup

    // 每个goroutine使用自己的会话副本，并在结束后关闭
    defer session.Close() 

    db := session.DB("mydb") // 从会话副本获取数据库实例

    fmt.Println("处理用户 - ID:", user.Id, " EMAIL:", user.Email)

    result := Post{}
    time.Sleep(50 * time.Millisecond) // 模拟耗时操作

    iter := db.C("posts").Find(bson.M{"user_id": user.Id}).Iter()

    for iter.Next(&result) {
        fmt.Println("  帖子 - ID:", result.Id, " 描述:", result.Description)
    }
    if err := iter.Close(); err != nil {
        fmt.Println("迭代器关闭错误:", err)
    }
}

func main() {
    masterSession, err := mgo.Dial("localhost:27017")
    if err != nil {
        panic(err)
    }
    defer masterSession.Close() // 确保主会话在所有goroutine完成后关闭

    db := masterSession.DB("mydb")

    // 清理旧数据并插入新数据
    db.C("users").DropCollection()
    db.C("posts").DropCollection()

    db.C("users").Insert(&User{Id: "user1", Email: "user1@example.com"})
    db.C("users").Insert(&User{Id: "user2", Email: "user2@example.com"})
    db.C("posts").Insert(&Post{Id: "post1_1", UserId: "user1", Description: "User1's first post"})
    db.C("posts").Insert(&Post{Id: "post1_2", UserId: "user1", Description: "User1's second post"})
    db.C("posts").Insert(&Post{Id: "post2_1", UserId: "user2", Description: "User2's first post"})

    fmt.Println("开始处理用户...")

    var wg sync.WaitGroup // 声明一个WaitGroup

    result := User{}
    iter := db.C("users").Find(nil).Iter()
    for iter.Next(&result) {
        wg.Add(1) // 每启动一个goroutine，计数器加1
        // 为每个goroutine创建一个会话副本
        go handleUser(masterSession.Copy(), &result, &wg) 
    }
    if err := iter.Close(); err != nil {
        fmt.Println("主迭代器关闭错误:", err)
    }

    wg.Wait() // 阻塞主函数，直到所有goroutine都调用了wg.Done()
    fmt.Println("所有用户和帖子处理完毕，主函数即将退出。")
}

// 示例伪代码，非完整实现
func main() {
    // ...
    done := make(chan struct{}, numUsers) // 创建一个带缓冲的channel
    for iter.Next(&result) {
        go func(user *User) {
            defer func() { done <- struct{}{} }() // 完成后发送信号
            // handleUser 逻辑，同样需要 session.Copy()
        }(&result)
    }

    // 等待所有goroutine完成
    for i := 0; i < numUsers; i++ {
        <-done
    }
    // ...
}