记一次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)
}

总结：

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