mongo-drive 基础使用

mongo-drive.md

基础

package main

import (
    "context"
    "log"

    "go.mongodb.org/mongo-driver/bson"
    "go.mongodb.org/mongo-driver/bson/primitive"
    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/mongo/options"
    "go.mongodb.org/mongo-driver/mongo/readpref"
    "go.mongodb.org/mongo-driver/x/bsonx"
)

type Book struct {
    Id       primitive.ObjectID `bson:"_id"`
    Name     string
    Category string
    Weight   int
    Author   AuthorInfo
}

type AuthorInfo struct {
    Name    string
    Country string
}

const (
    categoryComputer = "计算机"
    categorySciFi    = "科幻"
    countryChina     = "中国"
    countryAmerica   = "美国"
)

var (
    books = []interface{}{
        &Book{
            Id:       primitive.NewObjectID(),
            Name:     "深入理解计算机操作系统",
            Category: categoryComputer,
            Weight:   1,
            Author: AuthorInfo{
                Name:    "兰德尔 E.布莱恩特",
                Country: countryAmerica,
            },
        },
        &Book{
            Id:       primitive.NewObjectID(),
            Name:     "深入理解Linux内核",
            Category: categoryComputer,
            Weight:   1,
            Author: AuthorInfo{
                Name:    "博韦，西斯特",
                Country: countryAmerica,
            },
        },
        &Book{
            Id:       primitive.NewObjectID(),
            Name:     "三体",
            Category: categorySciFi,
            Weight:   1,
            Author: AuthorInfo{
                Name:    "刘慈欣",
                Country: countryChina,
            },
        },
    }
)

func main() {
    log.SetFlags(log.Llongfile | log.LstdFlags)

    opts := options.Client().ApplyURI("mongodb://localhost:27017")

    // 连接数据库
    client, err := mongo.Connect(context.Background(), opts)
    if err != nil {
        log.Fatal(err)
    }

    // 判断服务是不是可用
    if err = client.Ping(context.Background(), readpref.Primary()); err != nil {
        log.Fatal(err)
    }

    // 获取数据库和集合
    collection := client.Database("mydb").Collection("book")

    // 清空文档
    err = collection.Drop(context.Background())
    if err != nil {
        log.Fatal(err)
    }

    // 设置索引
    idx := mongo.IndexModel{
        Keys:    bsonx.Doc{{"name", bsonx.Int32(1)}},
        Options: options.Index().SetUnique(true),
    }
    idxRet, err := collection.Indexes().CreateOne(context.Background(), idx)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.Indexes().CreateOne:", idxRet)

    // 插入一条数据
    insertOneResult, err := collection.InsertOne(context.Background(), books[0])
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.InsertOne: ", insertOneResult.InsertedID)

    // 插入多条数据
    insertManyResult, err := collection.InsertMany(context.Background(), books[1:])
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.InsertMany: ", insertManyResult.InsertedIDs)

    // 获取数据总数
    count, err := collection.CountDocuments(context.Background(), bson.D{})
    if err != nil {
        log.Fatal(count)
    }
    log.Println("collection.CountDocuments:", count)

    // 查询单条数据
    var one Book
    err = collection.FindOne(context.Background(), bson.M{"name": "三体"}).Decode(&one)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.FindOne: ", one)

    // 查询多条数据(方式一)
    cur, err := collection.Find(context.Background(), bson.D{})
    if err != nil {
        log.Fatal(err)
    }
    if err := cur.Err(); err != nil {
        log.Fatal(err)
    }
    var all []*Book
    err = cur.All(context.Background(), &all)
    if err != nil {
        log.Fatal(err)
    }
    cur.Close(context.Background())

    log.Println("collection.Find curl.All: ", all)
    for _, one := range all {
        log.Println(one)
    }

    // 查询多条数据(方式二)
    cur, err = collection.Find(context.Background(), bson.D{})
    if err != nil {
        log.Fatal(err)
    }
    if err := cur.Err(); err != nil {
        log.Fatal(err)
    }
    for cur.Next(context.Background()) {
        var b Book
        if err = cur.Decode(&b); err != nil {
            log.Fatal(err)
        }
        log.Println("collection.Find cur.Next:", b)
    }
    cur.Close(context.Background())

    // 模糊查询
    cur, err = collection.Find(context.Background(), bson.M{"name": primitive.Regex{Pattern: "深入"}})
    if err != nil {
        log.Fatal(err)
    }
    if err := cur.Err(); err != nil {
        log.Fatal(err)
    }
    for cur.Next(context.Background()) {
        var b Book
        if err = cur.Decode(&b); err != nil {
            log.Fatal(err)
        }
        log.Println("collection.Find name=primitive.Regex{深入}: ", b)
    }
    cur.Close(context.Background())

    // 二级结构体查询
    cur, err = collection.Find(context.Background(), bson.M{"author.country": countryChina})
    // cur, err = collection.Find(context.Background(), bson.D{bson.E{"author.country", countryChina}})
    if err != nil {
        log.Fatal(err)
    }
    if err := cur.Err(); err != nil {
        log.Fatal(err)
    }
    for cur.Next(context.Background()) {
        var b Book
        if err = cur.Decode(&b); err != nil {
            log.Fatal(err)
        }
        log.Println("collection.Find author.country=", countryChina, ":", b)
    }
    cur.Close(context.Background())

    // 修改一条数据
    b1 := books[0].(*Book)
    b1.Weight = 2
    update := bson.M{"$set": b1}
    updateResult, err := collection.UpdateOne(context.Background(), bson.M{"name": b1.Name}, update)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.UpdateOne:", updateResult)

    // 修改一条数据，如果不存在则插入
    new := &Book{
        Id:       primitive.NewObjectID(),
        Name:     "球状闪电",
        Category: categorySciFi,
        Author: AuthorInfo{
            Name:    "刘慈欣",
            Country: countryChina,
        },
    }
    update = bson.M{"$set": new}
    updateOpts := options.Update().SetUpsert(true)
    updateResult, err = collection.UpdateOne(context.Background(), bson.M{"_id": new.Id}, update, updateOpts)
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.UpdateOne:", updateResult)

    // 删除一条数据
    deleteResult, err := collection.DeleteOne(context.Background(), bson.M{"_id": new.Id})
    if err != nil {
        log.Fatal(err)
    }
    log.Println("collection.DeleteOne:", deleteResult)
}

索引

func createUniqueIndex(collection string, keys ...string) {
    db := DB.Mongo.Database(setting.DatabaseSetting.DBName).Collection(collection)
    opts := options.CreateIndexes().SetMaxTime(10 * time.Second)

    indexView := db.Indexes()
    keysDoc := bsonx.Doc{}
    
    // 复合索引
    for _, key := range keys {
        if strings.HasPrefix(key, "-") {
            keysDoc = keysDoc.Append(strings.TrimLeft(key, "-"), bsonx.Int32(-1))
        } else {
            keysDoc = keysDoc.Append(key, bsonx.Int32(1))
        }
    }

    // 创建索引
    result, err := indexView.CreateOne(
        context.Background(),
        mongo.IndexModel{
            Keys:    keysDoc,
            Options: options.Index().SetUnique(true),
        },
        opts,
    )
    if result == "" || err != nil {
        Logger.Error("EnsureIndex error", zap.String("error", err.Error()))
    }
}

查询

func FindNodes() ([]DBNode, error) {
    var nodes []DBNode

    c := Connect(setting.DatabaseSetting.DBName, superNodeC)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    opts := options.Find().SetSort(bsonx.Doc{{"vote_count", bsonx.Int32(-1)}})
    cursor, err := c.Find(ctx, M{}, opts)
    if err != nil {
        return nil, err
    }

    for cursor.Next(context.Background()) {
        var node DBNode
        if err = cursor.Decode(&node); err != nil {
            return nil, err
        } else {
            nodes = append(nodes, node)
        }
    }

    return nodes, nil
}

插入

func Insert(db, collection string, docs ...interface{}) (*mongo.InsertManyResult, error) {
    c := Connect(db, collection)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    return c.InsertMany(ctx, docs)
}


func InsertNode(node DBNode) error {
    _, err := Insert(setting.DatabaseSetting.DBName, superNodeC, node)
    return err
}

修改

func Update(db, collection string, query, update interface{}) (*mongo.UpdateResult, error) {
    c := Connect(db, collection)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    
    opts := options.Update().SetUpsert(true)
    return c.UpdateOne(ctx, query, update,opts)
}

删除

func Remove(db, collection string, query interface{}) (*mongo.DeleteResult, error) {
    c := Connect(db, collection)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    return c.DeleteOne(ctx, query)
}


func RemoveNode(pubKey string) error {
    findM := M{"pub_key": pubKey}
    _, err := Remove(setting.DatabaseSetting.DBName, superNodeC, findM)
    return err
}

事务

• 需要使用SessionContext
• 所有的修改之前需要查询的，请都使用SessionContext（即都使用事务）

因为多处使用，所以封装了一个方法；
在这个方法中需要实现的方法是Exec的operator

type DBTransaction struct {
    Commit func(mongo.SessionContext) error
    Run    func(mongo.SessionContext, func(mongo.SessionContext, DBTransaction) error) error
    Logger *logging.Logger
}

func NewDBTransaction(logger *logging.Logger) *DBTransaction {
    var dbTransaction = &DBTransaction{}
    dbTransaction.SetLogger(logger)
    dbTransaction.SetRun()
    dbTransaction.SetCommit()
    return dbTransaction
}

func (d *DBTransaction) SetCommit() {
    d.Commit = func(sctx mongo.SessionContext) error {
        err := sctx.CommitTransaction(sctx)
        switch e := err.(type) {
        case nil:
            d.Logger.Info("Transaction committed.")
            return nil
        default:
            d.Logger.Error("Error during commit...")
            return e
        }
    }
}

func (d *DBTransaction) SetRun() {
    d.Run = func(sctx mongo.SessionContext, txnFn func(mongo.SessionContext, DBTransaction) error) error {
        err := txnFn(sctx, *d) // Performs transaction.
        if err == nil {
            return nil
        }
        d.Logger.Error("Transaction aborted. Caught exception during transaction.",
            zap.String("error", err.Error()))

        return err
    }
}

func (d *DBTransaction) SetLogger(logger *logging.Logger) {
    d.Logger = logger
}

func (d *DBTransaction) Exec(mongoClient *mongo.Client, operator func(mongo.SessionContext, DBTransaction) error) error {
    ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute)
    defer cancel()

    return mongoClient.UseSessionWithOptions(
        ctx, options.Session().SetDefaultReadPreference(readpref.Primary()),
        func(sctx mongo.SessionContext) error {
            return d.Run(sctx, operator)
        },
    )
}



//具体调用
func SyncBlockData(node models.DBNode) error {
    dbTransaction := db_session_service.NewDBTransaction(Logger)

    // Updates two collections in a transaction.
    updateEmployeeInfo := func(sctx mongo.SessionContext, d db_session_service.DBTransaction) error {
        err := sctx.StartTransaction(options.Transaction().
            SetReadConcern(readconcern.Snapshot()).
            SetWriteConcern(writeconcern.New(writeconcern.WMajority())),
        )
        if err != nil {
            return err
        }
        err = models.InsertNodeWithSession(sctx, node)
        if err != nil {
            _ = sctx.AbortTransaction(sctx)
            d.Logger.Info("caught exception during transaction, aborting.")
            return err
        }

        return d.Commit(sctx)
    }

    return dbTransaction.Exec(models.DB.Mongo, updateEmployeeInfo)
}

分页查询

func Find(database *mongo.Database,collection string,limit,index int64) (data []map[string]interface,err error){
	ctx, cannel := context.WithTimeout(context.Background(), time.Minute)
	defer cannel()
	var findoptions *options.FindOptions
	if limit > 0 {
		findoptions = &options.FindOptions{}
		findoptions.SetLimit(limit)
		findoptions.SetSkip(limit * index)
	}
	cur, err := database.Collection(collection).Find(ctx, bson.M{}, findoptions)
	if err != nil {
		return nil, err
	}
	defer cur.Close(context.Background())
	err = cur.All(context.Background(), &data)
	return
}

补充.md

转载 https://github.com/leonguo/go

mongodb在golang的应用

定义结构体

type User struct {
	Username string "user_name"
	Age      int    "age"
}
users := []User{}

查询 find

• 查找一个

selecter := bson.M{"_id":"test"}
err = s.DB("test").C("counters").Find(selecter).One(&result)

• 根据objectId查找

err = s.DB("test").C("users").FindId("58d9024569ad41655482e5df").One(&users)

• 查找多个

selecter = bson.M{"user_name":"ggg"}
iter := s.DB("test").C("users").Find(selecter).Sort("_id").Iter()
for iter.Next(&users) {
    fmt.Printf("users name: %v age: %v\n",users.Username,users.Age)
}

• 单条件查找 等于=

selecter := bson.M{"user_name":"ggg"}
err = s.DB("test").C("users").Find(selecter).All(&users)
fmt.Printf("%v", users)

• 单条件查找 不等于!= $ne

selecter := bson.M{"user_name":bson.M{"$ne":"ggg"}}

• 单条件查找 >($gt) >=($gte) <=($lte) <(lt) eq相等 ne、neq不相等， gt大于， lt小于 gte、ge大于等于 lte、le 小于等于 not非 mod求模

selecter := bson.M{"age":bson.M{"$gte":"10"}}    

• 单条件查找 in

selecter := bson.M{"user_name":bson.M{"$in":[]{"ggg","ggg1"}}}

• 多条件查询 and

selecter := bson.M{"user_name":"ggg","age":20}

• 多条件查询 or

selecter := bson.M{"$or":[]bson.M{bson.M{"user_name":"ggg"},bson.M{"age":20}}}

• 分组查询

match := bson.M{"user_name" : "ggg"}
group := bson.M{
        "_id":  nil,
        "user_name" :bson.M{"$last":"$type"},
        "age" :bson.M{"$last":"$type"},
        "total_num" :bson.M{"$sum":1},
        "avg_age" :bson.M{"$avg":"$age"},
    }
sort :=  bson.M{
        "age": -1,
    }
user := User{}
pipeline := []bson.M{{"$match":match},{"$group":group},{"$sort":sort}}
iter := s.DB("test").C("users").Pipe(pipeline).Iter()
for iter.Next(&user) {
    fmt.Printf("users name: %v age: %v\n",user.Username,user.AvgAge)
}

管道增加group可对分组后的数据统计

新增 insert

selecter := bson.M{"user_name":"ggg","age":20}
err = s.DB("test").C("users").Insert(selecter)

更新 update

• 更新（$set）增加数量($inc) 数组增加一个元素($push) 数组删除一个元素($pull)

selecter = bson.M{"user_name":"ggg"}
update := bson.M{"$set":bson.M{"age":21}}
err = s.DB("test").C("users").Update(selecter, update)

删除 delete

selecter = bson.M{"user_name":"ggg1"}
info, err := s.DB("test").C("users").RemoveAll(selecter)

简单查询操作符（非聚合操作）

• (1）$eq : 用来等值条件过滤某一个key的值。 用法示例（过滤某个key等于某个值，可以用 $eq) db.op_test.find({"name":{$eq:"steven"}})

• 2）$gt : 用来判断某个key值大于某个指定的值。

  用法示例（过滤某个key值，大于某个指定的值） db.op_test.find({"age":{$gt:19}})

• 3）$gt : 用来判断某个key值大于等于某个指定的值。

用法示例（过滤某个key值，大于某个指定的值）
db.op_test.find({"age":{$gte:19}})

• 4）$lt : 用来判断某个key值小于某个指定的值。

用法示例（过滤某个key值，小于某个指定的值）
db.op_test.find({"age":{$lt:20}})

• 5）$lte : 用来判断某个key值小于等于某个指定的值。

用法示例（过滤某个key值，小于等于某个指定的值）
db.op_test.find({"age":{$lte:20}})

• 6）$ne : 用来不等值条件过滤某一个key的值。

用法示例（过滤某个key不等于某个值，可以用 $ne)
db.op_test.find({"name":{$ne:"steven"}})

• 7）$in : 用来指定某个key的值在指定的离散的值域内

用法示例（过滤某个key的值是否符合条件，和$or的区别就是，or可以组合不同key的不同条件,而$in则只的是单一条件）
db.op_test.find({"name":{$in:["steven","jack"]}})

• 8）$nin : 和$in相对，用来指定key值不存在某个指定的离散值域内。

用法示例（和$in的用法相反）
db.op_test.find({"name":{$nin:["steven","jack"]}})
逻辑（Logic）运算相关

• 1） $or : 任意组合不同的查询条件（可以针对任意key的限制条件），只要满足任意组合条件中的一个即可。 用法示例（返回 name 为 steven 或者 age 等于 20 的文档）： db.op_test.find({"$or" : [{"name":"steven"},{"age":20}]})
• 2） $and: 和$or操作符相对，任意组合不同的查询条件（可以针对任意key的限制条件），并且必须同时满足所有条件。

用法示例（返回 name 为 steven 并且 age 等于 20 的文档）：
db.op_test.find({"$and" : [{"name":"steven"},{"age":20}]})

• 3） $not： 元条件语句，需要和其他条件语句组合使用。

用法示例（$not 和 $lt 操作符组合使用，返回 age 大于等于20的文档）：
db.op_test.find({"age":{"$not":{"$lt":20}}})

• 4） $nor：和$or相反，表示所有条件均不能满足则返回。

用法示例（凡是 name 为 steven 或者 age 为 20 的全部过滤掉）：
db.op_test.find({"$nor" : [{"name":"steven"},{"age":20}]})
元素(Element)运算相关

• 1） $exists: 查询不包含某一个属性（key）的文档。

用法实例（凡是包含name这个key的文档全部返回）
db.op_test.find({"name":{"$exists":true}})

用法实例（凡是不包含name这个key的文档全部返回）
db.op_test.find({"name":{"$exists":false}})

ps：true和false的区别就是判断是否包含这个key

• 2） $type : 过滤某个字段是某一个BSON数据类型的数据。

用法示例（返回所有name字段为String类型的所有文档）
db.op_test.find({"name":{"$type":2}})
ps：name后面的数字具体查询列表参见：http://docs.mongodb.org/manual/reference/operator/query/type/#op._S_type
求值（Evaluation）操作相关

• 1） $mod : 取余操作符，筛选经过区域操作后，结果符合条件的文档。

用法示例(返回age的值和 4 求余后 结果为 0 的数据)
db.op_test.find({"age" : {"$mod" : [4,0]}})

• 2） $regex : 筛选值满足正则表达式的文档。

用法示例（返回 name 符合指定正则的数据，option选项限定正则的形式）
db.op_test.find({"name" : {$regex:"stev*",$options:"i"}})
ps：options相关参见：http://docs.mongodb.org/manual/reference/operator/query/regex/#op._S_regex

• 3） $text: 针对建立了全文索引的字段，实施全文检索匹配。

用法示例（针对构建全文索引的字段进行搜索，默认为英文，不支持中文）
db.op_test.find({"$text":{$search:"steven",$language:"en"}})
ps：目前支持的语言以及缩写，参见：http://docs.mongodb.org/manual/reference/text-search-languages/

• 4） $where: 强大的查询关键字，但性能较差，可以传入js表达式或js函数来筛选数据。

用法示例（返回满足传入的js函数的文档，函数表示文档中只要任意字段的值为"steven"则返回）
db.op_test.find({"$where":function(){
    for(var index in this) {
        if(this[index] == "steven") {
            return true;
        }
    }
    return false;
}})
数组（Array）相关操作
$all : 数组查询操作符，查询条件是一个数组，被查询的字段也是一个数组，要求被查询的数组类型的字段要是查询条件数组的超集（即大数组要包含小数组）。

用法示例：(查询key value对应的数组值，要同时包含"a","b","c"三个元素)
db.op_test.find({"values":{$all:["a","b"]}})
$elemMatch : 数组查询操作符，用来指定数组的每一个元素同时满足所罗列的条件，如果不指定，则条件会是或的关系

用法示例：(要匹配 values数组中，至少有一个元素，满足所有的条件)
用于指定嵌套文档操作，具体事例参见：http://docs.mongodb.org/manual/reference/operator/projection/elemMatch/
$size: 用于某个数组类型的key对应值的数量满足要求。

用法示例：筛选出来包含数组元素个数为3的文档。
db.op_test.find({"values":{$size : 3}})
评论（Comments）相关操作
$comment: 在查询、更新或其他操作执行过程中，可以通过添加$comment操作符添加评论。改评论会被记录在日志中，用于后续分析。

用法示例
db.collection.find( { <query>, $comment: <comment> } )
地理位置（Geospatial）相关操作
$geoWithin: 这个操作符基于2d 空间索引，首先要针对文档的某个字段建立一个2d的空间索引，然后利用此操作符，可以在一个2d空间范围内指定一个多变形，$geoWithin操作符就是查询出包含在多变形范围内的点。

详见：http://docs.mongodb.org/manual/reference/operator/query/geoWithin/#op._S_geoWithin
$geoIntersects: 同样基于2d空间索引，计算当前的空间范围和指定的geo多变形取交集。

详见：http://docs.mongodb.org/manual/reference/operator/query/geoIntersects/#op._S_geoIntersects
$near ：基于2d空间索引，指定一个点，返回该点有近及远的所有的点。

详见：http://docs.mongodb.org/manual/reference/operator/query/near/#op._S_near
$nearSphere: 基于2d空间索引，指定一个点，由近及远的返回所有的点，和$near操作符不同的是计算距离的方式 $nearSphere计算的是球面距离。$near计算的是坐标距离。

投影相关操作
$ : 对你们看错，就只是一个$操作符，如果文档中某个value是数组类型，通过 $ 操作符可以指定数组字段的投影，返回数组字段中第一个匹配的那个元素，相当于截断了原来的整个数组，只返回第一个值。

用法示例：(会返回values数组中，第一个和"a"相等的元素，也就是返回"a")
db.op_test.find({"values":{$eq:"a"}},{"values.$":1})
返回结果如下：
{ "_id" : ObjectId("551117417cbfa0a55db5c5b9"), "values" : [ "a" ] }
$elemMatch : 这个操作符上面数组操作有涉及，其另外一个效果就是，在嵌套文档的应用中，返回数组中第一个符合条件的文档，可以限定多种组合条件。

详见：http://docs.mongodb.org/manual/reference/operator/projection/elemMatch/#proj._S_elemMatch
$meta ： 和全文索引 text index 组合使用，针对一个带有全文索引的元素，指定改操作符，可以返回和查询条件相似的分数，分数越高，匹配度越高。

用法示例：
db.op_test.find({"$text":{$search:"steven",$language:"en"}},{score:{$meta:"textScore"}})
执行结果：
{ "_id" : ObjectId("550fdba3c118f1b20bd51a9f"), "name" : "steven", "age" : 20, "score" : 1.1 }
$slice : 数组类型字段的投影操作，返回原来数据的一个子集.针对一个数组，其有如下几种返回子集的方式：

用法示例： 返回博客的前10条评论
db.blog.find({"comments":{"$slice":10}})

用法示例： 返回博客的后10条评论
db.blog.find({"comments":{"$slice":10}})

用法示例： 返回博客跳过前10条，然后返回第11 ~ 15条
db.blog.find({"comments":{"$slice":[10,5]}})

nice job