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

}