Golang Mgo步调

Question

我正在写一个快速写入MongoDB的应用程序。 mongodb和mgo处理太快。我的问题是，有没有办法让我确定mongo无法跟上并开始阻止？但我也不想不必要地阻止。 这里是模拟问题的代码示例：

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()您的会话。

要回答你的问题，这可能是一个完美的渠道使用案例。在一个goroutine中将这些项目加入到通道中，并将它们消耗掉/将它们写入另一个Mongo中。您可以调整频道的大小以满足您的需求。生产者线程在尝试发送通道时会阻塞该通道。

您可能还想玩Safe()方法设置。将W设置为0将会使Mongo处于一种“随意丢失”的模式，这将显着加速性能，并有可能丢失一些数据。您也可以更改超时时间。

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) 
    } 

}