mongodb - GoLang MongoDB 连接泄漏

Question

您好

这是一个代码示例，每秒从 MongoDB 执行检索。 我的问题是每次检索都会打开额外的连接(根据 MongoDB 日志)

代码:

package main

import (
  "os"
  "fmt"
  "gopkg.in/mgo.v2"
  "time"
  "gopkg.in/mgo.v2/bson"
  )

const (
    host1       = "localhost"
    port1       = "27017"
    dbName      = "test_db"
    collectionName  = "TEST_COLLECTION"
)
type Data struct {
    InternalId   bson.ObjectId `bson:"_id" json:"_id,omitempty"`
    Key1         string
    Key2         string
    Key3         int64
}


func main() {

   fmt.Println("Starting mongo worker ... ")
   finished := make(chan bool)
   go DoWorkerJob(finished)
   <-finished
}


func DoWorkerJob(finished chan bool) {

  session, err := GetSession()
  defer     session.Close()

  if err != nil {
        panic(err)
        os.Exit(1)
   }


 for {
    fmt.Println("Retrieving data ...")
    collection := GetCollection(*session,collectionName)

    allData, err := GetAllRows(collection)
    if err != nil {
        panic(err)
        continue
    }

    if allData != nil {

        fmt.Println("Total retrieved: ", len(allData), " documents.")
    }


    time.Sleep(time.Duration(1000) * time.Millisecond)
  }

   finished <- true
}


 func GetAllRows(collection *mgo.Collection) ([]Data, error) {
   var results []Data

   err := collection.Find(nil).All(&results)
   if err != nil {
       panic(err)
       return nil, err
   }
   return results, nil
 }

func GetSession() (*mgo.Session, error) {
  fmt.Println("Creating session ...")
  MongoDBHosts1 := host1 + ":" + port1

  mongoDBDialInfo := &mgo.DialInfo{
      Addrs:    []string{MongoDBHosts1},
      Timeout:  5 * time.Second,
      Database: dbName,
  }

  mongoSession, err := mgo.DialWithInfo(mongoDBDialInfo)

  if err != nil {
      panic(err)
      return nil, err
  }

  mongoSession.SetSocketTimeout(5 * time.Second)
  mongoSession.SetMode(mgo.Monotonic, true)

  session := mongoSession.New()

  fmt.Println("Session created!")
  return session, nil
 }


func GetCollection(session mgo.Session, queueName string) (*mgo.Collection) 
{

  fmt.Println("Creating collection ...")
  collection := session.Copy().DB(dbName).C(queueName)
  fmt.Println("Collection created!")
  return collection

 }

程序输出:

Starting mongo worker ... 
Creating session ...
Session created!
Retrieving data ...
Creating collection ...
Collection created!
Total retrieved:  3  documents.
Retrieving data ...
Creating collection ...
Collection created!
Total retrieved:  3  documents.
Retrieving data ...
Creating collection ...
Collection created!
Total retrieved:  3  documents.
Retrieving data ...

MongoDB 日志:

2017-08-03T11:24:53.600+0300 I NETWORK  [initandlisten] waiting for connections on port 27017
2017-08-03T11:25:38.785+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54591 #1 (1 connection now open)
2017-08-03T11:25:41.952+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54592 #2 (2 connections now open)
2017-08-03T11:25:45.260+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54593 #3 (3 connections now open)
2017-08-03T11:26:19.327+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54594 #4 (4 connections now open)
2017-08-03T11:26:38.797+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54595 #5 (5 connections now open)
2017-08-03T11:26:41.964+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54596 #6 (6 connections now open)
2017-08-03T11:26:45.269+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54597 #7 (7 connections now open)
2017-08-03T11:27:19.338+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54599 #8 (8 connections now open)
2017-08-03T11:38:37.106+0300 I NETWORK  [initandlisten] connection accepted from 127.0.0.1:54836 #9 (9 connections now open)

我在这里做错了什么？ Mongo 中的连接数达到数千，最终将其杀死......

Answer 1

每次复制 session 时，都应该将其关闭。

重写您的 GetCollection 和 GetAllRows，使用一个函数，例如:

func FetchData(session mgo.Session, queueName string) ([]Data, error) {
  fmt.Println("Creating collection ...")
  sess := session.Copy()
  collection := sess.DB(dbName).C(queueName)
  fmt.Println("Collection created!")

  defer  sess.Close() 

   var results []Data

   err := collection.Find(nil).All(&results)
   if err != nil {
       panic(err)
       return nil, err
   }
   return results, nil
}

注意线

 defer  sess.Close()