澳门新葡亰赌995577如何设计并实现一个线程安全的 Map ?(下篇)

澳门新葡亰赌995577 2

背景:

澳门新葡亰赌995577 1

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

在上篇中,我们已经讨论过如何去实现一个 Map
了,并且也讨论了诸多优化点。在下篇中,我们将继续讨论如何实现一个线程安全的
Map。说到线程安全,需要从概念开始说起。

问题

澳门新葡亰赌995577 2

大概是这样

线程安全就是如果你的代码块所在的进程中有多个线程在同时运行,而这些线程可能会同时运行这段代码。如果每次运行结果和单线程运行的结果是一样的,而且其他的变量的值也和预期的是一样的,就是线程安全的。

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)
}
  1. 可重入 Re-entrancy

相信大家都能看出问题:当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了。

  1. 线程本地存储

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

如果变量已经被本地化,所以每个线程都有自己的私有副本。这些变量通过子程序和其他代码边界保留它们的值,并且是线程安全的,因为这些变量都是每个线程本地存储的,即使访问它们的代码可能被另一个线程同时执行,依旧是线程安全的。

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

  1. 不可变量
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)
}

对象一旦初始化以后就不能改变。这意味着只有只读数据被共享,这也实现了固有的线程安全性。可变操作可以通过为它们创建新对象,而不是修改现有对象的方式去实现。
Java,C#和Python 中的字符串的实现就使用了这种方法。

但是这样还是有问题,相信经验比较丰富的老哥一眼就能看出来,问题处在socketInter,
ok =
pushList.sockets.Load(ip)这行代码上,如果map中没有这个值,且有多个协程同时访问到这行代码,显然这几个协程的ok都会置为false,然后都进入第一个if代码块,创建多个socket实例,并且争相覆盖原有值。

第一类方法都比较简单,通过代码改造就可以实现。但是如果遇到一定要进行线程中共享数据的情况,第一类方法就解决不了了。这时候就出现了第二类解决方案,利用线程同步的方法来解决线程安全问题。

单纯解决这个问题也很简单,就是使用sync.Map.LoadOrStore(key interface{}, value interface{}) (v interface{}, loaded bool)这个api,来原子地去做读写。

今天就从线程同步开始说起。