出现错误时重新创建 mgo 会话（读取 tcp 127.0.0.1:46954->127.0.0.1:27017: i/o timeout）

Question

我想知道 Go 中使用 mgo 的 MongoDB 会话管理，尤其是关于如何正确确保会话关闭以及如何对写入失败做出反应。

我已阅读以下内容：

Best practice to maintain a mgo session

Should I copy session for each operation in mgo?

仍然无法将其应用于我的情况。

我有两个 goroutine 将一个又一个事件存储到 MongoDB 中，共享同一个 *mgo.Session，两者看起来都像下面这样：

func storeEvents(session *mgo.Session) {
    session_copy := session.Copy()
    // *** is it correct to defer the session close here? <-----
    defer session_copy.Close()
    col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
    for {
        event := GetEvent()
        err := col.Insert(&event)
        if err != nil {
            // *** insert FAILED - how to react properly? <-----
            session_copy = session.Copy()
            defer session_copy.Close()
        }
    }
}

col.Insert(&event) 几个小时后返回错误

read tcp 127.0.0.1:46954->127.0.0.1:27017: i/o timeout

我不确定如何对此做出反应。发生此错误后，它会发生在所有后续写入中，因此似乎我必须创建一个新会话。对我来说似乎有替代方案：

1) 重启整个 goroutine，即

if err != nil {
    go storeEvents(session)
    return
}

2) 创建一个新的会话副本

if err != nil {
    session_copy = session.Copy()
    defer session_copy.Close()
    col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
    continue
}

--> 我如何使用defer session_copy.Close() 是否正确？ （注意上面的 defer 引用了另一个会话的 Close() 函数。无论如何，这些会话永远不会关闭，因为该函数永远不会返回。也就是说，随着时间的推移，许多会话将被创建而不是关闭。

其他选项？

Answer 1

所以我不知道这是否会对您有所帮助，但我对此设置没有任何问题。

我有一个从中导入的 mongo 包。这是我的 mongo.go 文件的模板

package mongo

import (
    "time"

    "gopkg.in/mgo.v2"
)

var (
    // MyDB ...
    MyDB DataStore
)

// create the session before main starts
func init() {
    MyDB.ConnectToDB()
}

// DataStore containing a pointer to a mgo session
type DataStore struct {
    Session *mgo.Session
}

// ConnectToTagserver is a helper method that connections to pubgears' tagserver
// database
func (ds *DataStore) ConnectToDB() {
    mongoDBDialInfo := &mgo.DialInfo{
        Addrs:    []string{"ip"},
        Timeout:  60 * time.Second,
        Database: "db",
    }
    sess, err := mgo.DialWithInfo(mongoDBDialInfo)
    if err != nil {
        panic(err)
    }
    sess.SetMode(mgo.Monotonic, true)
    MyDB.Session = sess
}

// Close is a helper method that ensures the session is properly terminated
func (ds *DataStore) Close() {
    ds.Session.Close()
}

然后在另一个包中，例如 main 根据下面的评论更新

package main

import (
    "../models/mongo"
)

func main() {
    // Grab the main session which was instantiated in the mongo package init function
    sess := mongo.MyDB.Session
    // pass that session in
    storeEvents(sess)
}

func storeEvents(session *mgo.Session) {
    session_copy := session.Copy()
    defer session_copy.Close()

    // Handle panics in a deferred fuction
    // You can turn this into a wrapper (middleware)
    // remove this this function, and just wrap your calls with it, using switch cases
    // you can handle all types of errors
    defer func(session *mgo.Session) {
        if err := recover(); err != nil {
            fmt.Printf("Mongo insert has caused a panic: %s\n", err)
            fmt.Println("Attempting to insert again")
            session_copy := session.Copy()
            defer session_copy.Close()
            col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
            event := GetEvent()
            err := col.Insert(&event)
            if err != nil {
                fmt.Println("Attempting to insert again failed")
                return
            }
            fmt.Println("Attempting to insert again succesful")
        }
    }(session)

    col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
    event := GetEvent()
    err := col.Insert(&event)
    if err != nil {
        panic(err)
    }
}

我在 AWS 上的生产服务器上使用了类似的设置。我每小时插入超过 100 万次。希望这可以帮助。我为确保 mongo 服务器可以处理连接所做的另一件事是在我的生产机器上创建 ulimit。在这个stack中讲到了