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

Question

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

我阅读了以下内容：

维持mgo会议的最佳实践

我应该为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()
        }
    }
}

上校插页(

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
}

--

还有其他选择吗？

Answer 1

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

我有一个mongo软件包，我是从那里进口的。这是我的mongo的模板。归档

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。在这一堆中讨论过