一日一学_Go语言mgo(mongo场景应用)

原文作者;WuXiao_ 来源:简书

本文重点介绍mgo使用,仅简单介绍mongodb。

mongodb特性

mongdb简单介绍

注意: 上图已经告知我们mongo不支持事务,在开发项目应用时,想要保证数据的完整性请考虑关系型数据库(经典例子银行转账)。 mongo提供了许多原子操作,比如文档的保存,修改,删除等,都是原子操作。所谓原子操作就是要么这个文档保存到mongodb,要么没有保存到mongodb,不会出现查询到的文档不完整的情况。

mgo简介

mgo 是 mongodb 的 GO 语言驱动包。 mgo官网:http://labix.org/mgo

mgo使用

mgo方案一
 1package mgo
 2
 3import (
 4    "flag"
 5    "gopkg.in/mgo.v2"
 6    "log"
 7    "study/conf"
 8)
 9
10var session *mgo.Session
11var database *mgo.Database
12
13func init() {
14     /*配置mongodb的josn文件,配置内容如下: 
15      {
16        "hosts": "localhost",
17        "database": "user"
18       }*/
19    filename := flag.String("config", "./conf/config.json", "Path to configuration file")
20    flag.Parse()
21    config := &conf.ConfigurationDatabase{}
22    config.Load(*filename)
23    var err error
24
25    dialInfo := &mgo.DialInfo{
26        Addrs:     []string{config.Hosts},
27        Direct:    false,
28        Timeout:   time.Second * 1,
29        PoolLimit: 4096, // Session.SetPoolLimit    }
30    //创建一个维护套接字池的session
31    session, err = mgo.DialWithInfo(dialInfo)
32
33    if err != nil {
34        log.Println(err.Error())
35    }
36    session.SetMode(mgo.Monotonic, true)
37    //使用指定数据库
38    database = session.DB(config.Database)
39
40}
41
42func GetMgo() *mgo.Session {
43    return session
44}
45
46func GetDataBase() *mgo.Database {
47    return database
48}
49
50func GetErrNotFound() error {
51    return mgo.ErrNotFound
52}

这里的 session 能够和 mongodb 集群中的所有Server通讯。

session设置的模式分别为:
  • Strong session 的读写一直向主服务器发起并使用一个唯一的连接,因此所有的读写操作完全的一致。
  • Monotonic session 的读操作开始是向其他服务器发起(且通过一个唯一的连接),只要出现了一次写操作,session 的连接就会切换至主服务器。由此可见此模式下,能够分散一些读操作到其他服务器,但是读操作不一定能够获得最新的数据。
  • Eventual session 的读操作会向任意的其他服务器发起,多次读操作并不一定使用相同的连接,也就是读操作不一定有序。session 的写操作总是向主服务器发起,但是可能使用不同的连接,也就是写操作也不一定有序。
 1//个人项目部分代码
 2type User struct {
 3    ID       bson.ObjectId `bson:"_id"`
 4    UserName string        `bson:"username"`
 5    Summary  string        `bson:"summary"`
 6    Age      int           `bson:"age"`
 7    Phone    int           `bson:"phone"`
 8    PassWord string        `bson:"password"`
 9    Sex      int           `bson:"sex"`
10    Name     string        `bson:"name"`
11    Email    string        `bson:"email"`
12}
13
14func Register(password string, username string) (err error) {
15    con := mgo.GetDataBase().C("user")
16    //可以添加一个或多个文档
17    /* 对应mongo命令行
18       db.user.insert({username:"13888888888",summary:"code",
19       age:20,phone:"13888888888"})*/
20    err = con.Insert(&User{ID: bson.NewObjectId(), UserName: username, PassWord: password})
21    return
22}
23
24func FindUser(username string) (User, error) {
25    var user User
26    con := mgo.GetDataBase().C("user")
27    //通过bson.M(是一个map[string]interface{}类型)进行
28    //条件筛选,达到文档查询的目的
29    /* 对应mongo命令行
30      db.user.find({username:"13888888888"})*/
31    if err := con.Find(bson.M{"username": username}).One(&user); err != nil {
32        if err.Error() != mgo.GetErrNotFound().Error() {
33            return user, err
34        }
35
36    }
37    return user, nil
38}

通过find()可以进行单个或者全部的查询,并且可以进行分页处理。下面为简单代码展示: con.Find(nil).Limit(5).Skip(0).All(&user)

 1package models
 2
 3import (
 4    "gopkg.in/mgo.v2/bson"
 5    "study/library/mgo"
 6    "time"
 7)
 8
 9type Diary struct {
10    Uid        bson.ObjectId `bson:"uid"`
11    ID         bson.ObjectId `bson:"_id"`
12    CreatTime  time.Time     `bson:"creattime"`
13    UpdateTime time.Time     `bson:"updatetime"`
14    Title      string        `bson:"title"`
15    Content    string        `bson:"content"`
16    Mood       int           `bson:'Mood"`
17    Pic        []string      `bson:'pic'`
18}
19
20//通过uid查找本作者文章,并且显示文章作者名字
21func FindDiary(uid string) ([]interface{}, error) {
22    con := mgo.GetDataBase().C("diary")
23       // 其中的lookup功能可以实现类似于mysql中的join操作,方便于关联查询。
24       /*对应mongo命令行
25        db.diary.aggregate([{$match:{uid: ObjectId("58e7a1b89b5099fdc585d370")}},
26        {$lookup{from:"user",localField:"uid",foreignField:"_id",as:"user"}},
27        {$project:{"user.name":1,title:1,content:1,mood:1}}]).pretty() 
28       */
29    pipeline := []bson.M{
30        bson.M{"$match": bson.M{"uid": bson.ObjectIdHex(uid)}},
31        bson.M{"$lookup": bson.M{"from": "user", "localField": "uid", "foreignField": "_id", "as": "user"}},
32        bson.M{"$project": bson.M{"user.name": 1, "title": 1, "content": 1, "mood": 1, "creattime": 1}},
33    }
34    pipe := con.Pipe(pipeline)
35    var data []interface{}
36    err := pipe.All(&data)
37    if err != nil {
38        return nil, err
39    }
40    return data, nil
41
42}
43
44func ModifyDiary(id, title, content string) (err error) {
45        con := mgo.GetDataBase().C("diary")
46         //更新
47        /*对应mongo命令行
48          db.diary.update({_id:ObjectId("58e7a1b89b5099fdc585d370")},
49           {$set:{title:"modify title",content:"modify content",
50           updatetime:new Date()})*/
51        err = con.Update(bson.M{"_id": id}, bson.M{"$set": bson.M{"title": title, "content": content, "updatetime": time.Now().Add(8 * time.Hour)}})
52        return
53
54}

mgo更新方法很多,如批量更新con.UpdateAll(selector, update),更新或插入数据con.Upsert(selector, update)

思路一会儿

mgo方案二

思考: session 会被全局使用,当在实际的程序中,我们可以开启goroutine 来处理每个连接,多个goroutine 可以通过 session.Clone() 来创建或复用连接,使用完成之后通过 session.Close() 来关闭这个连接。当并发很高时,看起来可以提高效率。

下面部分代码修改 :

 1import (
 2    "flag"
 3    "gopkg.in/mgo.v2"
 4    "log"
 5    "study/conf"
 6)
 7
 8var session *mgo.Session
 9var config *conf.ConfigurationDatabase
10
11func init() {
12     filename := flag.String("config", "./conf/config.json", "Path to configuration file")
13    flag.Parse()
14
15    config = &conf.ConfigurationDatabase{}
16    config.Load(*filename)
17    var err error
18
19    dialInfo := &mgo.DialInfo{
20        Addrs:     []string{config.Hosts},
21        Direct:    false,
22        Timeout:   time.Second * 1,
23        PoolLimit: 4096, // Session.SetPoolLimit
24    }
25    session, err = mgo.DialWithInfo(dialInfo)
26
27    if err != nil {
28        log.Println(err.Error())
29    }
30    session.SetMode(mgo.Monotonic, true)
31
32
33
34}
35
36type SessionStore struct {
37    session *mgo.Session
38}
39
40//获取数据库的collection
41func (d * SessionStore) C(name string) *mgo.Collection {
42    return d.session.DB(config.Database).C(name)
43}
44
45//为每一HTTP请求创建新的DataStore对象
46func New SessionStore() * SessionStore {
47    ds := & SessionStore{
48        session: session.Copy(),
49    }
50    return ds
51}
52
53func (d * SessionStore) Close() {
54    d.session.Close()
55}
56
57func GetErrNotFound() error {
58    return mgo.ErrNotFound
59}

对查找进行了修改

 1func FindUser(username string) (User, error) {
 2    var user User
 3    ds := mgo.NewSessionStore()
 4    defer ds.Close()
 5    con := ds.C("user")
 6    if err := con.Find(bson.M{"username": username}).One(&user); err != nil {
 7        if err.Error() != mgo.GetErrNotFound().Error() {
 8            return user, err
 9        }
10
11    }
12    return user, nil
13}

mgo方案一和二测试:

使用boom进行并发测试,并在每个 goroutine 里面sleep 5秒,这样是让连接暂时不释放,就可以看到 mgo 方案二 会不断创建新连接,方案一不会创建新连接。可以使用mongo shell 的db.serverStatus().connections来查看连接数。

mgo方案一测试连接数: 1000 并发:mongo 3个连接 5000 并发:mongo 3个连接。

mgo方案二测试连接数: 1000 并发:mongo 500多个连接 5000 并发:mongo 1400多个连接。

提示: mgo 默认连接池是 4096,在高并发下,如果每个 session都不调用 close(),会导致连接数会很快就达到 4096,并堵死其他请求,所以在使用clone() 或 copy()时 session 时一定要使用 defer close() 把连接关闭。启用 maxPoolLimit 参数会限制总连接大小,当连接超过限制总数当前协程 等待,直到可以创建连接。

测试结果:mgo方案一和方案二在并发下,效率差不多。

为什么

可能性,由于数据少或者处理的单个mongo无法看出效果。 由于目前自己项目只使用了一个mongo,后期使用多个mongo进行或在大量数据下测试。如果大家有什么好的建议,提出来进行学习思考。 推荐学习: http://goinbigdata.com/how-to-build-microservice-with-mongodb-in-golang/ 官方博客详讲了mgo并发处理,如下: https://www.mongodb.com/blog/post/running-mongodb-queries-concurrently-with-go


版权申明:内容来源网络,版权归原创者所有。除非无法确认,我们都会标明作者及出处,如有侵权烦请告知,我们会立即删除并表示歉意。谢谢。

原文发布于微信公众号 - Golang语言社区(Golangweb)

原文发表时间:2019-01-04

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券