Go语言mgo使用情况

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

mongodb特性

mongdb简单介绍


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

mgo简介

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

mgo使用

mgo方案一
  1.  
    package mgo
  2.  
     
  3.  
    import (
  4.  
    "flag"
  5.  
    "gopkg.in/mgo.v2"
  6.  
    "log"
  7.  
    "study/conf"
  8.  
    )
  9.  
     
  10.  
    var session *mgo.Session
  11.  
    var database *mgo.Database
  12.  
     
  13.  
    func 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.  
     
  42.  
    func GetMgo() *mgo.Session {
  43.  
    return session
  44.  
    }
  45.  
     
  46.  
    func GetDataBase() *mgo.Database {
  47.  
    return database
  48.  
    }
  49.  
     
  50.  
    func GetErrNotFound() error {
  51.  
    return mgo.ErrNotFound
  52.  
    }
  53.  
     

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

session设置的模式分别为:
  • Strong
    session 的读写一直向主服务器发起并使用一个唯一的连接，因此所有的读写操作完全的一致。
  • Monotonic
    session 的读操作开始是向其他服务器发起（且通过一个唯一的连接），只要出现了一次写操作，session 的连接就会切换至主服务器。由此可见此模式下，能够分散一些读操作到其他服务器，但是读操作不一定能够获得最新的数据。
  • Eventual
    session 的读操作会向任意的其他服务器发起，多次读操作并不一定使用相同的连接，也就是读操作不一定有序。session 的写操作总是向主服务器发起，但是可能使用不同的连接，也就是写操作也不一定有序。
  1.  
    //个人项目部分代码
  2.  
    type 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.  
     
  14.  
    func 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.  
     
  24.  
    func 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.  
    }
  39.  
     
  40.  
     

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

  1.  
    package models
  2.  
     
  3.  
    import (
  4.  
    "gopkg.in/mgo.v2/bson"
  5.  
    "study/library/mgo"
  6.  
    "time"
  7.  
    )
  8.  
     
  9.  
    type 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查找本作者文章，并且显示文章作者名字
  21.  
    func 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.  
     
  44.  
    func 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.  
    }
  55.  
     
  56.  
     
  57.  
     

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

 
思路一会儿
mgo方案二

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

下面部分代码修改 :

  1.  
    import (
  2.  
    "flag"
  3.  
    "gopkg.in/mgo.v2"
  4.  
    "log"
  5.  
    "study/conf"
  6.  
    )
  7.  
     
  8.  
    var session *mgo.Session
  9.  
    var config *conf.ConfigurationDatabase
  10.  
     
  11.  
    func 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.  
     
  36.  
    type SessionStore struct {
  37.  
    session *mgo.Session
  38.  
    }
  39.  
     
  40.  
    //获取数据库的collection
  41.  
    func (d * SessionStore) C(name string) *mgo.Collection {
  42.  
    return d.session.DB(config.Database).C(name)
  43.  
    }
  44.  
     
  45.  
    //为每一HTTP请求创建新的DataStore对象
  46.  
    func New SessionStore() * SessionStore {
  47.  
    ds := & SessionStore{
  48.  
    session: session.Copy(),
  49.  
    }
  50.  
    return ds
  51.  
    }
  52.  
     
  53.  
    func (d * SessionStore) Close() {
  54.  
    d.session.Close()
  55.  
    }
  56.  
     
  57.  
    func GetErrNotFound() error {
  58.  
    return mgo.ErrNotFound
  59.  
    }
  60.  
     

对查找进行了修改

  1.  
    func 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.  
    }
  14.  
     
  15.  
     

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

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/ExMan/p/13936073.html