【整理】golang mgo连接池实现

由于mgo连接池是自带的，你只需要使用session.Copy()拿到一个复制的session，用完之后session.Close()即可。

代码实现如下：

package db
import (
    "labix.org/v2/mgo"
    //  "labix.org/v2/mgo/bson"
)
const (
    USER string = "user"
    MSG  string = "msg"
)
var (
    session      *mgo.Session
    databaseName = "test"
)
func Session() *mgo.Session {
    if session == nil {
        var err error
        session, err = mgo.Dial("localhost")
        if err != nil {
            panic(err) // no, not really
        }
    }
    return session.Clone()
}
func M(collection string, f func(*mgo.Collection)) {
    session := Session()
    defer func() {
        session.Close()
        if err := recover(); err != nil {
            Log("M", err)
        }
    }()
    c := session.DB(databaseName).C(collection)
    f(c)
}

使用方式：

var m = M{}
db.M(db.USER, func(c *mgo.Collection) {
        c.Find(M{"_id": id}).One(m)
        Log("find one m", m, m["name"], m["img"])
    })

参考：http://golangtc.com/t/53073b69320b526197000067

另外参考该链接：golang mgo的mongo连接池设置：必须手动加上maxPoolSize

该博客中所写的内容，其实大家在使用mgo的连接用完注意close掉即可避免上文中的堵塞问题。

最后，附送上mongodb官方提供的示例代码，参考网址：https://www.mongodb.com/blog/post/running-mongodb-queries-concurrently-with-go

// This program provides a sample application for using MongoDB with
// the mgo driver.
package main

import (
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"log"
	"sync"
	"time"
)

const (
	MongoDBHosts = "ds035428.mongolab.com:35428"
	AuthDatabase = "goinggo"
	AuthUserName = "guest"
	AuthPassword = "welcome"
	TestDatabase = "goinggo"
)

type (
	// BuoyCondition contains information for an individual station.
	BuoyCondition struct {
		WindSpeed     float64 `bson:"wind_speed_milehour"`
		WindDirection int     `bson:"wind_direction_degnorth"`
		WindGust      float64 `bson:"gust_wind_speed_milehour"`
	}

	// BuoyLocation contains the buoy's location.
	BuoyLocation struct {
		Type        string    `bson:"type"`
		Coordinates []float64 `bson:"coordinates"`
	}

	// BuoyStation contains information for an individual station.
	BuoyStation struct {
		ID        bson.ObjectId `bson:"_id,omitempty"`
		StationId string        `bson:"station_id"`
		Name      string        `bson:"name"`
		LocDesc   string        `bson:"location_desc"`
		Condition BuoyCondition `bson:"condition"`
		Location  BuoyLocation  `bson:"location"`
	}
)

// main is the entry point for the application.
func main() {
	// We need this object to establish a session to our MongoDB.
	mongoDBDialInfo := &mgo.DialInfo{
		Addrs:    []string{MongoDBHosts},
		Timeout:  60 * time.Second,
		Database: AuthDatabase,
		Username: AuthUserName,
		Password: AuthPassword,
	}

	// Create a session which maintains a pool of socket connections
	// to our MongoDB.
	mongoSession, err := mgo.DialWithInfo(mongoDBDialInfo)
	if err != nil {
		log.Fatalf("CreateSession: %s\n", err)
	}

	// Reads may not be entirely up-to-date, but they will always see the
	// history of changes moving forward, the data read will be consistent
	// across sequential queries in the same session, and modifications made
	// within the session will be observed in following queries (read-your-writes).
	// http://godoc.org/labix.org/v2/mgo#Session.SetMode
	mongoSession.SetMode(mgo.Monotonic, true)

	// Create a wait group to manage the goroutines.
	var waitGroup sync.WaitGroup

	// Perform 10 concurrent queries against the database.
	waitGroup.Add(10)
	for query := 0; query < 10; query++ {
		go RunQuery(query, &waitGroup, mongoSession)
	}

	// Wait for all the queries to complete.
	waitGroup.Wait()
	log.Println("All Queries Completed")
}

// RunQuery is a function that is launched as a goroutine to perform
// the MongoDB work.
func RunQuery(query int, waitGroup *sync.WaitGroup, mongoSession *mgo.Session) {
	// Decrement the wait group count so the program knows this
	// has been completed once the goroutine exits.
	defer waitGroup.Done()

	// Request a socket connection from the session to process our query.
	// Close the session when the goroutine exits and put the connection back
	// into the pool.
	sessionCopy := mongoSession.Copy()
	defer sessionCopy.Close()

	// Get a collection to execute the query against.
	collection := sessionCopy.DB(TestDatabase).C("buoy_stations")

	log.Printf("RunQuery : %d : Executing\n", query)

	// Retrieve the list of stations.
	var buoyStations []BuoyStation
	err := collection.Find(nil).All(&buoyStations)
	if err != nil {
		log.Printf("RunQuery : ERROR : %s\n", err)
		return
	}

	log.Printf("RunQuery : %d : Count[%d]\n", query, len(buoyStations))
}

文章的脚注信息由WordPress的wp-posturl插件自动生成

打赏

微信扫一扫，打赏作者吧～