mongodb go封装

mongodb-go-driver驱动的封装：package Dalimport ("fmt""github.com/mongodb/mongo-go-driver/mongo""context""time""github.com/mongodb/mongo-go-driver/bson""github.com/mongodb/mongo-go-driver/mo...

henreash

3924人浏览 · 2019-02-01 22:02:57

henreash · 2019-02-01 22:02:57 发布

mongodb-go-driver驱动的封装：

package Dal

import (
	"fmt"
	"github.com/mongodb/mongo-go-driver/mongo"
	"context"
	"time"
	"github.com/mongodb/mongo-go-driver/bson"
	"github.com/mongodb/mongo-go-driver/mongo/readpref"
	"Model"
)

type MongoUtils struct {
	Con *mongo.Client
	Db *mongo.Database
	ServerIp string
	Port int
}

func (o *MongoUtils) OpenConn() (con *mongo.Client){
	connString := fmt.Sprintf("mongodb://%s:%d", o.ServerIp, o.Port)
	_, err := url.Parse(connString)
	if err != nil{
		fmt.Println(err)
		return
	}
	opts:= &options.ClientOptions{}
	opts.SetAuth(options.Credential{AuthMechanism:"SCRAM-SHA-1", AuthSource:"AuthDb", Username:"henreash", Password:"XXXXXX"})
	opts.SetMaxPoolSize(5)//设置连接池大小
	ctx, _ := context.WithTimeout(context.Background(), 30*time.Second)
	con, err = mongo.Connect(ctx, connString, opts)
	if err != nil{
		fmt.Println(err)
		return nil
	}
	err = con.Ping(ctx, readpref.Primary())
	if err != nil{
		fmt.Println(err)
		return nil
	}
	o.Con = con
	return con
}

func (o *MongoUtils) SetDb(db string)  {
	if o.Con ==nil{
		panic("连接为空...")
	}
	o.Db = o.Con.Database(db)
}

func (o *MongoUtils) FindOne(col string, filter bson.M) (bson.M, error) {
	if o.Db == nil || o.Con == nil{
		return nil, fmt.Errorf("没有初始化连接和数据库信息！")
	}
	table := o.Db.Collection(col)
	ctx,_ := context.WithTimeout(context.Background(), 5*time.Second)
	var result bson.M
	err := table.FindOne(ctx, filter).Decode(&result)
	if err != nil{
		return nil, err
	}
	return result, nil
}

func (o *MongoUtils) FindMore(col string, filter bson.M) ([]bson.M, error){
	if o.Db == nil || o.Con == nil{
		return nil, fmt.Errorf("没有初始化连接和数据库信息！")
	}
	table := o.Db.Collection(col)
	ctx,_ := context.WithTimeout(context.Background(), 5*time.Second)
	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
	cur, err2 := table.Find(ctx, filter)
	if err2 != nil {
		fmt.Print(err2)
		return nil, err2
	}
	defer cur.Close(ctx)
	var resultArr []bson.M
	for cur.Next(ctx){
		var result bson.M
		err3 := cur.Decode(&result)
		if err3 != nil {
			return nil, err3
		}
		resultArr = append(resultArr, result)
	}
	return resultArr, nil
}

func Bson2Odj(val interface{}, obj interface{})(error)  {
	data,err := bson.Marshal(val)
	if err != nil{
		return err
	}
	bson.Unmarshal(data, obj)
	return nil
}

connect方法的连接选项如下，牵涉到连接池、连接超时、可选服务器、证书等配置信息，非常灵活：

	opts := options.Client().SetAppName("foo").SetAuth(options.Credential{
		AuthMechanism:           "MONGODB-X509",
		AuthMechanismProperties: map[string]string{"foo": "bar"},
		AuthSource:              "$external",
		Password:                "supersecurepassword",
		Username:                "admin",
	}).SetConnectTimeout(500 * time.Millisecond).SetHeartbeatInterval(15 * time.Second).SetHosts([]string{
		"mongodb://localhost:27018",
		"mongodb://localhost:27019",
	}).SetLocalThreshold(time.Second).SetMaxConnIdleTime(30 * time.Second).SetMaxPoolSize(150).
		SetReadConcern(rc).SetReadPreference(rp).SetReplicaSet("foo").
		SetRetryWrites(retryWrites).SetServerSelectionTimeout(time.Second).
		SetSingle(false).SetSocketTimeout(2 * time.Second).SetSSL(&options.SSLOpt{
		Enabled:                      true,
		ClientCertificateKeyFile:     "client.pem",
		ClientCertificateKeyPassword: nil,
		Insecure:                     false,
		CaFile:                       "ca.pem",
	}).SetWriteConcern(wc)

增删改查：

type Role struct {
	RoleId int `json:"RoleId" bson:"RoleId"`
	RoleName string `json:"RoleName" bson:"RoleName"`
	Valid string `json:"Valid" bson:"Valid"`
	Remark string `json:"Remark" bson:"Remark"`
}

type RoleSlice struct {
	RoleList []Role
} 

package Bll

import (
	"encoding/json"
	"Dal"
	"Model"
	"context"
	"github.com/globalsign/mgo/bson"
	"fmt"
	"github.com/mongodb/mongo-go-driver/mongo"
)

type bllRole struct {
}

func BllRoleTest(){
	//insertRole(3, "软件工程师", "true", "")
	//updateRole(3, "软件工程师1", "true", "更新了")
	removeRole(3)
}

func updateRole(roleId int, roleName string, valid string, remark string) string {
	utils := Dal.MongoUtils{
		ServerIp: "localhost",
		Port:     27017,
		}
	utils.OpenConn()
	defer utils.Con.Disconnect(context.Background())//连接用完后记得要释放
	utils.SetDb("AuthDb")
	col := utils.Db.Collection("Role")
	utils.Con.UseSession(context.Background(), func(ses mongo.SessionContext) error {
		ses.StartTransaction()
		_,err := col.UpdateOne(context.Background(), bson.M{"RoleId":roleId},bson.M{"$set": bson.M{"RoleId":roleId, "RoleName":roleName, "Valid":valid, "Remark":remark}})
		if err != nil {
			ses.AbortTransaction(ses)
			return err
		}
		ses.CommitTransaction(ses)
		return nil
	})
	return "suc"
}

func insertRole(roleId int, roleName string, valid string, remark string) string {
	utils := Dal.MongoUtils{
		ServerIp: "localhost",
		Port:     27017,}
	utils.OpenConn()
	defer utils.Con.Disconnect(context.Background())
	utils.SetDb("AuthDb")
	col := utils.Db.Collection("Role")
	suc:=true
	utils.Con.UseSession(context.Background(), func(ses mongo.SessionContext) error {
		ses.StartTransaction()
		_,err:=col.InsertOne(context.Background(), bson.M{"RoleId":roleId, "RoleName":roleName, "Valid":valid, "Remark":remark})
		if err!=nil{
			suc = false
			ses.AbortTransaction(ses)
			fmt.Println(err)
			return err
		}
		ses.CommitTransaction(ses)
		return nil
	})

	if !suc{
		return "err"
	}
	result := Model.Role{
		RoleId:roleId,
		RoleName:roleName,
		Valid:valid,
		Remark:remark,
	}
	b, _ := json.Marshal(result)
	res := string(b)
	return res
}

func removeRole(roleId int) string {
	utils := Dal.MongoUtils{
		ServerIp: "localhost",
		Port:     27017,}
	utils.OpenConn()
	defer utils.Con.Disconnect(context.Background())
	utils.SetDb("AuthDb")
	col := utils.Db.Collection("Role")
	utils.Con.UseSession(context.Background(), func(ses mongo.SessionContext) error {
		ses.StartTransaction()
		_, err := col.DeleteOne(context.Background(), bson.M{"RoleId": roleId})
		if err != nil{
			ses.AbortTransaction(ses)
			return err
		}
		ses.CommitTransaction(ses)
		return nil
	})
	res := "suc"
	return res
}

数据操作的测试：

func MongoTest2(){
	util := MongoUtils{
		ServerIp:"localhost",
		Port:27017,
	}
	util.OpenConn()
        defer utils.Con.Disconnect(context.Background())//连接用完后记得要释放
	util.SetDb("AuthDb")
	filter := bson.M{"RoleName": "管理员"}
	res, err:=util.FindOne("Role", filter)
	if err != nil{
		fmt.Println(err)
	}
	result2:=Model.Role{}
	/*data, _ :=bson.Marshal(&res)
	bson.Unmarshal(data, &result2)*/
	Bson2Odj(&res, &result2)
	println(result2.RoleName)

	res2,err:=util.FindMore("Role", bson.M{})
	if err != nil{
		fmt.Println(err)
	}
	for i := 0; i < len(res2); i++{
		b := res2[i]
		result3 := Model.Role{}
		/*d,_:=bson.Marshal(&b)
		bson.Unmarshal(d,&result3)*/
		Bson2Odj(&b, &result3)
		fmt.Println(result3.RoleName)
	}
}

func MongoTest() {
	client, err := mongo.NewClient("mongodb://localhost:27017")
	if err != nil{
		fmt.Print(err)
	}
	ctx, _ := context.WithTimeout(context.Background(), 30*time.Second)
	errC := client.Connect(ctx)
	if errC != nil{
		fmt.Print(errC)
	}
	errP := client.Ping(ctx, readpref.Primary())
	if errP != nil{
		fmt.Print(errP)
	}

	session, _:= client.StartSession()
	session.StartTransaction()
	db := client.Database("AuthDb")
	table := db.Collection("Role")

	filter := bson.M{"RoleName": "管理员"}
	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
	var result1 bson.M
	err = table.FindOne(ctx, filter).Decode(&result1)
	if err != nil {
		fmt.Println(err)
	}
	//bson 2 struct
    result2:=Model.Role{}
    data, _ :=bson.Marshal(&result1)
    bson.Unmarshal(data, &result2)
	println(result2.RoleName)
    //struct 2 bson
	data, _ = bson.Marshal(&result2)
	bson2 := bson.M{}
	bson.Unmarshal(data, &bson2)
	println(bson2["RoleName"])
    //table.FindOne(ctx,filter).Decode(&result2)

    var filter2 bson.M
	ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
	cur, err2 := table.Find(ctx, filter2)
	if err2 != nil {
		//log.Fatal(err2)
		fmt.Print(err2)
	}
	defer cur.Close(ctx)
	for cur.Next(ctx){
		var result bson.M
		err3 := cur.Decode(&result)
		if err3 != nil {
			fmt.Print(err3)
		}
		fmt.Print(result["RoleId"])
		fmt.Print(result["RoleName"])
	}

	/*ctx, _ = context.WithTimeout(context.Background(), 5*time.Second)
	res, errI := table.InsertOne(ctx, bson.M{"RoleId":2, "RoleName":"333", "Valid":"false"})
	if errI !=nil{
		fmt.Print(errI)
	}
	fmt.Print(res.InsertedID)*/
	//事务
	colT := db.Collection("TmpCollection")
	ctx = context.Background()
	colT.InsertOne(ctx, bson.M{"_id":"222", "name":"ddd","age":10})
	db.Client().UseSession(ctx, func(ses mongo.SessionContext) error{
		err = ses.StartTransaction()
		if err != nil {
			fmt.Println(err)
			return err
		}
		colT := db.Collection("TmpCollection")
		_,err = colT.InsertOne(ses, bson.M{"_id":"333", "name":"ddd","age":10})
		if err!=nil{
			fmt.Println(err)
			return err
		}
		_,err = colT.InsertOne(ses, bson.M{"_id":"222", "name":"ddd","age":10})
		if err!=nil{
			//与上面提交的_id:222记录主键冲突，抛出异常，调用下面的回滚操作
			fmt.Println(err)
			ses.AbortTransaction(ses)
		}
		ses.CommitTransaction(ses)
		return nil
	})
}

结论:Mongodb的数据增删改比关系型数据库操作要方便，查询略复杂。

创建用户和验证：

     1）首先启动mongod，不带--auth选项，在mongo下创建超级用户：
        db=db.getSiblingDB('admin')
        db.createUser({user:"AdminUser",pwd:"password",roles:["userAdminAnyDatabase"])
    2) 加上--auth选项启动mongod服务，用刚创建的超级用户登陆，创建新的用户henreash，密码123qwe!@#
        在mongo下：
        use AuthDb
        db.auth("AdminUser","password")
        db.createUser({user:"henreash",pwd:"123qwe!@#",roles:["readWrite","dbAdmin"]

go中连接mongodb时需要指定用户名和密码，否则报权限验证错误。

func (o *MongoUtils) OpenConn() (con *mongo.Client){
	connString := fmt.Sprintf("mongodb://%s:%d", o.ServerIp, o.Port)
	_, err := url.Parse(connString)
	if err != nil{
		fmt.Println(err)
		return
	}
	opts:= &options.ClientOptions{}
	opts.SetAuth(options.Credential{AuthMechanism:"SCRAM-SHA-1", AuthSource:"AuthDb", Username:"henreash", Password:"XXXXXX"})
	ctx, _ := context.WithTimeout(context.Background(), 30*time.Second)
	con, err = mongo.Connect(ctx, connString, opts)
	if err != nil{
		fmt.Println(err)
		return nil
	}
	err = con.Ping(ctx, readpref.Primary())
	if err != nil{
		fmt.Println(err)
		return nil
	}
	o.Con = con
	return con
}

使用mongo-go-driver驱动执行关联查询（相对与sql的left join）：

func MongoTest3(){
	util := MongoUtils{
		ServerIp:"localhost",
		Port:27017,
	}
	util.OpenConn()
        defer utils.Con.Disconnect(context.Background())//连接用完后记得要释放
	util.SetDb("AuthDb")
	toConvert := `[
        {
            "$project": {
                "_id": 0,
                "A": "$$ROOT"
            }
        }, 
        {
            "$lookup": {
                "localField": "A.TypeId",
                "from": "RoleType",
                "foreignField": "_id",
                "as": "B"
            }
        }, 
        {
            "$unwind": {
                "path": "$B",
                "preserveNullAndEmptyArrays": true
            }
        }, 
        {
            "$project": {
                "A.RoleId": "$A.RoleId",
                "A.RoleName": "$A.RoleName",
                "A.Valid": "$A.Valid",
                "B.TypeName": "$B.TypeName"
            }
        }
    ]`
	table := util.Db.Collection("Role")
	pipeLine := mongo.Pipeline{}
	err := bson.UnmarshalExtJSON([]byte(toConvert), true, &pipeLine)
	if err != nil {
		fmt.Println(err)
	}
	var b = true
	options := options.AggregateOptions{AllowDiskUse:&b}
	cur,_ := table.Aggregate(context.Background(), pipeLine, &options)
	defer cur.Close(context.Background())
	for cur.Next(context.Background()){
		var result bson.M
		err3 := cur.Decode(&result)
		if err3 != nil {
			fmt.Print(err3)
		}
		a:=result["A"]
		b := result["B"]
		fmt.Printf("%s   %s\n", a.(bson.M)["RoleName"] ,b.(bson.M)["TypeName"])//interface{}类型转换为bson.M
	}
}

toConvert变量是从Studio 3T的sql关联查询针对Node.js翻译代码中拷贝的，即pipeline变量。很可惜Studio 3T还没有针对go自动生成关联查询代码，但Node.js、C#、Java、Python、Php都进行了翻译。在golang中可以进行简单的转换即可使用。

Mongo提供了一个查看连接、操作的工具，MongoStat.exe，启动后每秒显示一次当前服务器的连接和操作情况，从而可以查看我们打开的连接是否已经被关闭。从这个工具可以看到，mongo-go-driver驱动提供的池没有生效。而且Connect后一定要defer Disconnect。