golang:mgo剖析之Session
chase-wind · · 1525 次点击 · · 开始浏览golang操作mongo使用的包是"gopkg.in/mgo.v2",coding过程中需要并发读写mongo数据库,简单观摩了下源码,记录下自己的一些理解,如有错误,敬请斧正。
一般来说,我们直接这样创建一个session:
Session, err = mgo.Dial(URL)
if err != nil {
log.Println(err)
}
来看看Dial这个函数做了什么:
func Dial(url string) (*Session, error) {
session, err := DialWithTimeout(url, 10*time.Second)
if err == nil {
session.SetSyncTimeout(1 * time.Minute)
session.SetSocketTimeout(1 * time.Minute)
}
return session, err
}
调用DialWithTimeout函数设置默认的超时时间是10秒。该函数中调用了DialWithInfo这个函数,而DialWithInfo函数中比较重要是是调用了newSession,看看这个函数做了什么操作:
func newSession(consistency Mode, cluster *mongoCluster, timeout time.Duration) (session *Session) {
cluster.Acquire()
session = &Session{
cluster_: cluster,
syncTimeout: timeout,
sockTimeout: timeout,
poolLimit: 4096,
}
debugf("New session %p on cluster %p", session, cluster)
session.SetMode(consistency, true)
session.SetSafe(&Safe{})
session.queryConfig.prefetch = defaultPrefetch
return session
}
返回的session设置了一些默认的参数,暂时先忽略,直接看看Session的数据结构:
type Session struct {
m sync.RWMutex
cluster_ *mongoCluster
slaveSocket *mongoSocket
masterSocket *mongoSocket
slaveOk bool
consistency Mode
queryConfig query
safeOp *queryOp
syncTimeout time.Duration
sockTimeout time.Duration
defaultdb string
sourcedb string
dialCred *Credential
creds []Credential
poolLimit int
bypassValidation bool
}
m是mgo.Session的并发锁,因此所有的Session实例都是线程安全的。slaveSocket,masterSocket代表了该Session到mongodb主节点和从节点的一个物理连接的缓存。而Session的策略总是优先使用缓存的连接。是否缓存连接,由consistency也就是该Session的模式决定。假设在并发程序中,使用同一个Session实例,不使用Copy,而该Session实例的模式又恰好会缓存连接,那么,所有的通过该Session实例的操作,都会通过同一条连接到达mongodb。虽然mongodb本身的网络模型是非阻塞通信,请求可以通过一条链路,非阻塞地处理,但是会影响效率。
其次mgo.Session缓存的一主一从连接,实例本身不负责维护。也就是说,当slaveSocket,masterSocket任意其一,连接断开,Session自己不会重置缓存,该Session的使用者如果不主动重置缓存,调用者得到的将永远是EOF。这种情况在主从切换时就会发生,在网络抖动时也会发生。
mgo的DB句柄需要你做一个copy操作:
// Copy works just like New, but preserves the exact authentication
// information from the original session.
func (s *Session) Copy() *Session {
s.m.Lock()
scopy := copySession(s, true)
s.m.Unlock()
scopy.Refresh()
return scopy
}
copySession将源Session浅拷贝到临时Session中,这样源Session的配置就拷贝到了临时Session中。关键的Refresh,将源Session浅拷贝到临时Session的连接缓存指针,也就是slaveSocket,masterSocket置为空,这样临时Session就不存在缓存连接,而转为去尝试获取一个空闲的连接。
mgo自身维护了一套到mongodb集群的连接池。这套连接池机制以mongodb数据库服务器为最小单位,每个mongodb都会在mgo内部,对应一个mongoServer结构体的实例,一个实例代表着mgo持有的到该数据库的连接。看看这个连接池的定义:
type mongoServer struct {
sync.RWMutex
Addr string
ResolvedAddr string
tcpaddr *net.TCPAddr
unusedSockets []*mongoSocket
liveSockets []*mongoSocket
closed bool
abended bool
sync chan bool
dial dialer
pingValue time.Duration
pingIndex int
pingCount uint32
pingWindow [6]time.Duration
info *mongoServerInfo
}
info代表了该实例对应的数据库服务器在集群中的信息——是否master,ReplicaSetName等。而两个Slice,就是传说中的连接池。unusedSockets存储当前空闲的连接,liveSockets存储当前活跃中的连接,Session缓存的连接就同时存放在liveSockets切片中,而临时Session获取到的连接就位于unusedSockets切片中。
每个mongoServer都会隶属于一个mongoCluster结构,相当于mgo在内部,模拟出了mongo数据库集群的模型。
type mongoCluster struct {
sync.RWMutex
serverSynced sync.Cond
userSeeds []string
dynaSeeds []string
servers mongoServers
masters mongoServers
references int
syncing bool
direct bool
failFast bool
syncCount uint
setName string
cachedIndex map[string]bool
sync chan bool
dial dialer
}
mongoCluster持有一系列mongoServer的实例,以主从结构分散到两个数组中。 每个Session都会存储自己对应的,要操作的mongoCluster的引用。
前面的描述可以总结成下面这张图:
那么我们在使用的时候就可以创建一个Session,然后clone操作,用clone得到的copysession完成操作,结束后关闭这个copysession就可以了。
有疑问加站长微信联系(非本文作者)
入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889
关注微信- 请尽量让自己的回复能够对别人有帮助
- 支持 Markdown 格式, **粗体**、~~删除线~~、
`单行代码` - 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
- 图片支持拖拽、截图粘贴等方式上传
收入到我管理的专栏 新建专栏
golang操作mongo使用的包是"gopkg.in/mgo.v2",coding过程中需要并发读写mongo数据库,简单观摩了下源码,记录下自己的一些理解,如有错误,敬请斧正。
一般来说,我们直接这样创建一个session:
Session, err = mgo.Dial(URL)
if err != nil {
log.Println(err)
}
来看看Dial这个函数做了什么:
func Dial(url string) (*Session, error) {
session, err := DialWithTimeout(url, 10*time.Second)
if err == nil {
session.SetSyncTimeout(1 * time.Minute)
session.SetSocketTimeout(1 * time.Minute)
}
return session, err
}
调用DialWithTimeout函数设置默认的超时时间是10秒。该函数中调用了DialWithInfo这个函数,而DialWithInfo函数中比较重要是是调用了newSession,看看这个函数做了什么操作:
func newSession(consistency Mode, cluster *mongoCluster, timeout time.Duration) (session *Session) {
cluster.Acquire()
session = &Session{
cluster_: cluster,
syncTimeout: timeout,
sockTimeout: timeout,
poolLimit: 4096,
}
debugf("New session %p on cluster %p", session, cluster)
session.SetMode(consistency, true)
session.SetSafe(&Safe{})
session.queryConfig.prefetch = defaultPrefetch
return session
}
返回的session设置了一些默认的参数,暂时先忽略,直接看看Session的数据结构:
type Session struct {
m sync.RWMutex
cluster_ *mongoCluster
slaveSocket *mongoSocket
masterSocket *mongoSocket
slaveOk bool
consistency Mode
queryConfig query
safeOp *queryOp
syncTimeout time.Duration
sockTimeout time.Duration
defaultdb string
sourcedb string
dialCred *Credential
creds []Credential
poolLimit int
bypassValidation bool
}
m是mgo.Session的并发锁,因此所有的Session实例都是线程安全的。slaveSocket,masterSocket代表了该Session到mongodb主节点和从节点的一个物理连接的缓存。而Session的策略总是优先使用缓存的连接。是否缓存连接,由consistency也就是该Session的模式决定。假设在并发程序中,使用同一个Session实例,不使用Copy,而该Session实例的模式又恰好会缓存连接,那么,所有的通过该Session实例的操作,都会通过同一条连接到达mongodb。虽然mongodb本身的网络模型是非阻塞通信,请求可以通过一条链路,非阻塞地处理,但是会影响效率。
其次mgo.Session缓存的一主一从连接,实例本身不负责维护。也就是说,当slaveSocket,masterSocket任意其一,连接断开,Session自己不会重置缓存,该Session的使用者如果不主动重置缓存,调用者得到的将永远是EOF。这种情况在主从切换时就会发生,在网络抖动时也会发生。
mgo的DB句柄需要你做一个copy操作:
// Copy works just like New, but preserves the exact authentication
// information from the original session.
func (s *Session) Copy() *Session {
s.m.Lock()
scopy := copySession(s, true)
s.m.Unlock()
scopy.Refresh()
return scopy
}
copySession将源Session浅拷贝到临时Session中,这样源Session的配置就拷贝到了临时Session中。关键的Refresh,将源Session浅拷贝到临时Session的连接缓存指针,也就是slaveSocket,masterSocket置为空,这样临时Session就不存在缓存连接,而转为去尝试获取一个空闲的连接。
mgo自身维护了一套到mongodb集群的连接池。这套连接池机制以mongodb数据库服务器为最小单位,每个mongodb都会在mgo内部,对应一个mongoServer结构体的实例,一个实例代表着mgo持有的到该数据库的连接。看看这个连接池的定义:
type mongoServer struct {
sync.RWMutex
Addr string
ResolvedAddr string
tcpaddr *net.TCPAddr
unusedSockets []*mongoSocket
liveSockets []*mongoSocket
closed bool
abended bool
sync chan bool
dial dialer
pingValue time.Duration
pingIndex int
pingCount uint32
pingWindow [6]time.Duration
info *mongoServerInfo
}
info代表了该实例对应的数据库服务器在集群中的信息——是否master,ReplicaSetName等。而两个Slice,就是传说中的连接池。unusedSockets存储当前空闲的连接,liveSockets存储当前活跃中的连接,Session缓存的连接就同时存放在liveSockets切片中,而临时Session获取到的连接就位于unusedSockets切片中。
每个mongoServer都会隶属于一个mongoCluster结构,相当于mgo在内部,模拟出了mongo数据库集群的模型。
type mongoCluster struct {
sync.RWMutex
serverSynced sync.Cond
userSeeds []string
dynaSeeds []string
servers mongoServers
masters mongoServers
references int
syncing bool
direct bool
failFast bool
syncCount uint
setName string
cachedIndex map[string]bool
sync chan bool
dial dialer
}
mongoCluster持有一系列mongoServer的实例,以主从结构分散到两个数组中。 每个Session都会存储自己对应的,要操作的mongoCluster的引用。
前面的描述可以总结成下面这张图:
那么我们在使用的时候就可以创建一个Session,然后clone操作,用clone得到的copysession完成操作,结束后关闭这个copysession就可以了。