首页猿问 Go中的“显式关闭&r...

Go中的“显式关闭”mgo MongoDB

心有法竹 2021-11-08 16:49:51

我有一个运行 10 个 Web 服务（同类）的 Docker 集群。他们都在使用 MongoDB，其中包括数据持久性。这是在main()服务启动时从调用的代码：// Init establishes a connection with MongoDB instance.func Init(mongoURL string) *mgo.Session { mongo, err := mgo.Dial(mongoURL) misc.PanicIf(err) // make sure we are strongly consistent mongo.SetMode(mgo.Strong, true) // update global state db = mongo Entries = db.DB("").C("entries") Channels = db.DB("").C("channels") Settings = db.DB("").C("settings") Metadata = db.DB("").C("metadata") // only use this on first load, to confirm the settings are there // every refresh should be done via UpdateGlobalSettings (thread-safe) GlobalSettings = &GlobalSettingsStruct{} GlobalSettings.Init() return mongo}所以基本上API和工人只需使用全局变量，例如Entries，Settings等等。运行一段时间后，该服务停止正常工作。每蒙戈动作，比如Entries.find(...)回报的错误：Closed Explicitly。这意味着什么？我应该定期刷新 mongoDB 连接还是应该关闭它并重新建立与每个请求的连接？该应用程序以性能为导向，因此尽管 mongo 连接已关闭，但一切仍在运行，因为一切都在内存或集群缓存上运行。我不想做一些延迟处理的愚蠢事情。

查看完整描述

3 回答

12345678_0001

TA贡献1802条经验获得超5个赞

当sess.SetPoolLimit(50)不使用时mgo，在 10.000 个并发连接等压力下会发生许多错误。

当我限制池错误就消失了。

我在下面为这个问题创建了一个测试用例源代码，所以你可以很容易地在自己的机器上测试它。

你对这种行为有什么建议吗，我想听听。

package main

import (

"fmt"

"time"

// you can also use original go-mgo/mgo here as well

mgo "github.com/globalsign/mgo"

"github.com/globalsign/mgo/bson"

)

// TODO: put some records into db first:

// use testapi

// db.competitions.insert([

// {game_id: 1, game_name: "foo"},

// {game_id: 2, game_name: "bar"},

// {game_id: 3, game_name: "jazz"}

// ])

// NOTE: you might want to increase this depending on your machine power

// mine is:

// MacBook (Retina, 12-inch, Early 2015)

// 1,2 GHz Intel Core M

// 8 GB 1600 MHz DDR3

const ops = 10000

type m bson.M

func main() {

sess, err := mgo.DialWithTimeout("localhost", time.Second)

if err != nil {

panic(err)

}

defer sess.Close()

// NOTE: without this setting there are many errors

// see the end of the file

// setting pool limit prevents most of the timeouts

// sess.SetPoolLimit(50)

// sess.SetSocketTimeout(60 * time.Second)

sess.SetMode(mgo.Monotonic, true)

time.Sleep(time.Second)

done := make(chan bool, ops)

for i := 0; i < ops; i++ {

go func() {

defer func() { done <- true }()

result, err := fetchSomething(sess)

if err != nil {

fmt.Printf("ERR: %s\n", err)

}

fmt.Printf("RESULT: %+v\n", result)

}()

}

for i := 0; i < ops; i++ {

<-done

}

func fetchSomething(sess *mgo.Session) ([]m, error) {

s := sess.Copy()

defer s.Close()

result := []m{}

group := m{

"$group": m{

"_id": m{

"id": "$game_id",

"name": "$game_name",

}

project := m{

"$project": m{

"_id": "$_id.id",

"name": "$_id.name",

}

sort := m{

"$sort": m{

"_id": 1,

}

err := col(s, "competitions").Pipe([]m{group, project, sort}).All(&result)

if err != nil {

return nil, err

}

return result, nil

}

// col is a helper for selecting a column

func col(sess *mgo.Session, name string) *mgo.Collection {

return sess.DB("testapi").C(name)

}

ERRORS WITHOUT sess.SetPoolLimit(50)

$ go run socket_error.go

RESULT: [map[name:foo _id:1] map[_id:2 name:bar] map[_id:3 name:jazz]]

ERR: read tcp 127.0.0.1:52918->127.0.0.1:27017: read: connection reset by peer

ERR: write tcp 127.0.0.1:52084->127.0.0.1:27017: write: broken pipe

RESULT: []