记一次golang中sync.Map并发创建、读取的问题

背景:

我们有一个用go做的项目,其中用到了zmq4进行通信,一个简单的rpc过程,早期远端是使用一个map去做ip和具体socket的映射。

问题

大概是这样

struct SocketMap {
 sync.Mutex
 sockets map[string]*zmq4.Socket
}

然后调用的时候的代码大概就是这样的:

func (pushList *SocketMap) push(ip string, data []byte) {
 pushList.Lock()
 defer pushList.UnLock()
 socket := pushList.sockets[string]
 if socket == nil {
 socket := zmq4.NewSocket()
 //do some initial operation like connect
 pushList.sockets[ip] = socket
 }
 socket.Send(data)
}

相信大家都能看出问题:当push被并发访问的时候(事实上push会经常被并发访问),由于这把大锁的存在,同时只能有一个协程在临界区工作,效率是会被大大降低的。

解决方案:会带来crash的优化

所以我们决定使用sync.Map来替代这个设计,然后出了第一版代码,写的非常简单,只做了简单的替换:

struct SocketMap {
 sockets sync.Map
}
func (pushList *SocketMap) push(ip string, data []byte) {
 var socket *zmq4.Socket 
 socketInter, ok = pushList.sockets.Load(ip)
 if !ok {
 socket = zmq4.NewSocket()
 //do some initial operation like connect
 pushList.sockets.Store(ip, socket)
 } else {
 socket = socketInter.(*zmq4.Socket)
 }
 socket.Send(data)
}

乍一看似乎没什么问题?但是跑起来总是爆炸,然后一看log,提示有个非法地址。后来在github上才看到,zmq4.Socket不是线程安全的。上面的代码恰恰会造成多个线程同时拿到socket实例,然后就crash了。

解决方案2: 加一把锁也挡不住的冲突

然后怎么办呢?看来也只能加锁了,不过这次加锁不能加到整个map上,否则还会有性能问题,那就考虑减小锁的粒度吧,使用锁包装socket。这个时候我们的代码也就呼之欲出了:

struct SocketMutex{
 sync.Mutex
 socket *zmq4.Socket
}
struct SocketMap {
 sockets sync.Map
}
func (pushList *SocketMap) push(ip string, data []byte) {
 var socket *SocketMutex 
 socketInter, ok = pushList.sockets.Load(ip)
 if !ok {
 socket = &{
 socket: zmq4.NewSocket()
 }
 //do some initial operation like connect
 pushList.sockets.Store(ip, newSocket)
 } else {
 socket = socketInter.(*SocketMutex)
 }
 socket.Lock()
 defer socket.Unlock()
 socket.socket.Send(data)
}

但是这样还是有问题,相信经验比较丰富的老哥一眼就能看出来,问题处在socketInter, ok = pushList.sockets.Load(ip)这行代码上,如果map中没有这个值,且有多个协程同事访问到这行代码,显然这几个协程的ok都会置位false,然后都进入第一个if代码块,创建多个socket实例,并且争相覆盖原有值。
单纯解决这个问题也很简单,就是使用sync.Map.LoadOrStore(key interface{}, value interface{}) (v interface{}, loaded bool)这个api,来原子地去做读写。
然而这还没完,我们的写入新值的操作不光是调用一个api创建socket就完了,还要有一系列的初始化操作,我们必须保证在初始化完成之前,其他通过Load拿到这个实例的协程无法真正访问socket实例。
这时候显然sync.Map自带的机制已经无法解决这个问题了,那么我们必须寻求其他的手段,要么锁,要么就sync.WaitGroup或者whatever的其他什么东西。

解决方案3: 闭包带来的神奇体验

后来经大佬指点,我在encoder.go中看到了这么一段代码:

 346 func typeEncoder(t reflect.Type) encoderFunc { 
 347 if fi, ok := encoderCache.Load(t); ok { 
 348 return fi.(encoderFunc) 
 349 } 
 350 
 351 // To deal with recursive types, populate the map with an 
 352 // indirect func before we build it. This type waits on the 
 353 // real func (f) to be ready and then calls it. This indirect 
 354 // func is only used for recursive types. 
 355 var ( 
 356 wg sync.WaitGroup 
 357 f encoderFunc 
 358 ) 
 359 wg.Add(1) 
 360 fi, loaded := encoderCache.LoadOrStore(t, encoderFunc(func(e *encodeState, v reflect.Value, opts encOpts) {
 361 wg.Wait() 
 362 f(e, v, opts) 
 363 })) 
 364 if loaded { 
 365 return fi.(encoderFunc) 
 366 } 
 367 
 368 // Compute the real encoder and replace the indirect func with it. 
 369 f = newTypeEncoder(t, true) 
 370 wg.Done() 
 371 encoderCache.Store(t, f) 
 372 return f 
 373 } 

豁然开朗,我们可以在sync.Map中存放一个闭包函数,然后在闭包函数中等待本地的sync.WaitGroup完成再返回实例。于是最终的代码也就成型了。

struct SocketMutex{
 sync.Mutex
 socket *zmq4.Socket
}
struct SocketMap {
 sockets sync.Map
}
func (pushList *SocketMap) push(ip string, data []byte) {
 type SocketFunc func()*SocketMutex
 var (
 socket *SocketMutex
 w sync.WaitGroup
 )
 socket = &SocketMutex {
 socket : zmq4.NewSocket()
 } 
 w.Add(1)
 socketf, ok = pushList.sockets.LoadOrStore(ip, SocketFunc(func()*SocketMutex) {
 w.Wait()
 return socket
 })
 if !ok {
 socket = &{
 socket: zmq4.NewSocket()
 }
 //do some initial operation like connect
 w.Done()
 } else {
 socket = socketInter.(*SockeFunc)()
 }
 socket.Lock()
 defer socket.Unlock()
 socket.socket.Send(data)
}

总结:

并发代码中的竞争问题,每一行代码的重入性都要深思熟虑啊。
时间略晚,懒得总结了(逃

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传