Golang Mgo起搏

Question

我正在编写一个快速写入mongodb的应用程序。速度太快，无法处理mongodb和mgo。我的问题是，有什么办法让我确定芒果不能跟上并开始阻塞吗？但我也不想不必要地阻碍。下面是一个模拟问题的代码示例：

package main

import (
  "labix.org/v2/mgo"
  "time"
  "fmt"
)

// in database name is a string and age is an int

type Dog struct{
  Breed string "breed"
}

type Person struct{
  Name string "name"
  Pet Dog `bson:",inline"`
  Ts        time.Time
}

func insert(session *mgo.Session, bob Person){
  err := session.DB("db_log").C("people").Insert(&bob)
  if err != nil {
    panic("Could not insert into database")
  }
}

func main() {
  session, _ := mgo.Dial("localhost:27017")
  bob := Person{Name : "Robert", Pet : Dog{}}
  i := 0
  for {
    time.Sleep(time.Duration(1) * time.Microsecond)
    i++
    go insert(session, bob)
  }
}

我经常会犯这样的错误：

panic: Could not insert into database

或

panic: write tcp 127.0.0.1:27017: i/o timeout

Answer 1

我怀疑如果您allow Go to use multiple threads和Copy() then Close()您的会话，您将获得更好的性能。

要回答你的问题，这可能是一个完美的用例一个频道。在一个峡谷中将这些物品输入通道，然后将它们消耗/在另一个地方将它们写到Mongo。您可以调整频道的大小，以适应您的需要。一旦通道被填满，当它试图发送到它时，生产者线程将阻塞。

您还可能希望使用Safe()方法设置。设置W:0将使蒙戈进入“火与忘”模式，这将大大加快性能，冒着丢失一些数据的风险。您还可以更改超时时间。

Answer 2

我还没有进行测试，但我认为这段代码应该可以工作。我得到这个问题后，保留了很长一段时间，以便我有计时器更新会话每隔一段时间。

package main

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

// in database name is a string and age is an int

type Dog struct{
  Breed string "breed"
}

type Person struct{
  Name string "name"
  Pet Dog `bson:",inline"`
  Ts        time.Time
}

func insert(session *mgo.Session, bob Person){
  err := session.DB("db_log").C("people").Insert(&bob)
  if err != nil {
    panic("Could not insert into database")
  }
}

func main() {
  current_session, _ := mgo.Dial("localhost:27017")
  using_session := current_session
  bob := Person{Name : "Robert", Pet : Dog{}}

  /*
  * this technical to prevent connect timeout after long time connection on mongodb from golang session
  * Idea is simple: the session will be renew after certain time such as 1 hour
  */
  //ticker := time.NewTicker(time.Hour * 1)

  //Set 10 seconds for test
  ticker := time.NewTicker(time.Second * 10)

  go func() {

    for t := range ticker.C {
      fmt.Println("Tick at", t)
      new_session := current_session.Copy()
      fmt.Printf("Current session here %p\n", current_session)
      fmt.Printf("New session here %p\n", new_session)
      using_session = new_session
      //setTimeout 30 second before close old sesion, to make sure current instance use current connection isn't affect
      //time.AfterFunc(time.Second * 30, func() { 

      //Set 2 seconds for test
      time.AfterFunc(time.Second * 2, func() { 

        //close previous session

        current_session.Close()
        current_session = new_session

        //assign to new session

      })

    }
  }()

  i := 0
  for {
    time.Sleep(time.Duration(1) * time.Microsecond)
    i++
    go insert(using_session, bob)
  }

}