我有 2,000,000 條記錄的集合> db.events.count(); │2000000 我使用 golang mongodb 客戶端連接到數據庫package mainimport ( "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options")func main() { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() client, err := mongo.Connect(ctx, options.Client().ApplyURI("mongodb://localhost:27888").SetAuth(options.Credential{ Username: "mongoadmin", Password: "secret", })) if err != nil { panic(err) } defer func() { if err = client.Disconnect(ctx); err != nil { panic(err) } }() collection := client.Database("test").Collection("events") var bs int32 = 10000 var b = true cur, err := collection.Find(context.Background(), bson.D{}, &options.FindOptions{ BatchSize: &bs, NoCursorTimeout: &b}) if err != nil { log.Fatal(err) } defer cur.Close(ctx) s, n := runningtime("retrive db from mongo and publish to kafka") count := 0 for cur.Next(ctx) { var result bson.M err := cur.Decode(&result) if err != nil { log.Fatal(err) } bytes, err := json.Marshal(result) if err != nil { log.Fatal(err) } count++ msg := &sarama.ProducerMessage{ Topic: "hello", // Key: sarama.StringEncoder("aKey"), Value: sarama.ByteEncoder(bytes), } asyncProducer.Input() <- msg }但是該程序僅檢索大約 600,000 條記錄，而不是每次運行該程序時的 2,000,000 條。$ go run main.godonecount = 605426nErrors = 02020/09/18 11:23:43 End: retrive db from mongo and publish to kafka took 10.080603336s我不知道為什么？我想檢索所有 2,000,000 條記錄。謝謝你的幫助。
1 回答
慕娘9325324
TA貢獻1783條經驗 獲得超4個贊
您獲取結果的循環可能會提前結束，因為您使用相同的ctx上下文來迭代具有 10 秒超時的結果。
這意味著如果檢索和處理 200 萬條記錄（包括連接）時間超過 10 秒，上下文將被取消，因此游標也會報告錯誤。
請注意，設置FindOptions.NoCursorTimeout為true只是為了防止光標因不活動而超時，它不會覆蓋使用的上下文的超時。
使用另一個上下文來執行查詢并迭代結果，一個沒有超時的上下文，例如context.Background().
另請注意，要構造 的選項find，請使用輔助方法，因此它可能看起來像這樣簡單而優雅：
options.Find().SetBatchSize(10000).SetNoCursorTimeout(true)
所以工作代碼：
ctx2 := context.Background()
cur, err := collection.Find(ctx2, bson.D{},
options.Find().SetBatchSize(10000).SetNoCursorTimeout(true))
// ...
for cur.Next(ctx2) {
// ...
}
// Also check error after the loop:
if err := cur.Err(); err != nil {
log.Printf("Iterating over results failed: %v", err)
}
- 1 回答
- 0 關注
- 153 瀏覽
添加回答
舉報
0/150
提交
取消