Go语言中的锁与管道的运用

1.前言

2.锁解决方案

3.管道解决方案

4.总结

1.前言

在写H5小游戏的时候，由于需要对多个WebSocket连接进行增、删、查的管理和对已经建立连接的WebSocket通过服务端进行游戏数据交换的需求。于是定义了一个全局的map集合进行连接的管理，让所有的协程共享操作同一个map集合，进行各种WebSocket连接的操作。由于多个协程操作共享同一块内容，这时候就会遇到数据竞争和并发访问。

H5小游戏介绍：基于WebSocket通信的H5小游戏总结-CSDN博客

解决并发问题的常见方法有两种：

在结构体中增加 sync.RWMutex字段，每一个协程操作map集合的时候进行加锁操作，操作结束后进行解锁操作，保证同时只有一个协程操作map,避免并发问题。但是频繁的加锁和解锁操作会成为后期的性能瓶颈。
使用管道进行通信，由于管道本身就是线程安全的，所以我们在操作层面无需进行加锁和解锁操作，只需要另启一个协程进行管道的读取，如果有数据写入则进行map操作。我们在需要对map进行操作的时候向管道中写入数据即可。

由于第一次在项目中遇到并发问题，一开始没有意识到多个协程对同一个map进行操作需要保证线程安全。在老师查看代码后，说出map是线程不安全的时候，才意识到需要进行加锁操作或者其他方案来保证线程安全。

2.锁解决方案

第一版本的代码——加锁，解锁保证线程安全

在结构体中的ClientsMap进行操作的时候进行加锁和解锁的操作，保证线程安全。

// HupCenter ---使用锁，操作一个多线程共享的Map---//
type HupCenter struct {//第一个string-roomId 第二个string-userIdClientsMap map[string]map[string]*Client `json:"-"` mutex      sync.RWMutex
}// JoinHub  写操作 --将连接加入中心 前提RoomId不为空, 加入房间的时候需要检测当前房间里面的人数
func (h *HupCenter) JoinHub(c *Client) (flag bool) {h.mutex.Lock()defer h.mutex.Unlock()//先查询是否存在此一个roomId keyif myMap, ok := c.Hub.ClientsMap[c.User.RoomId]; ok { //有,加入房间//检测人数if len(myMap) == 1 {myMap[c.User.UserId] = cflag = true}} else { //没有,创建房间myMap := make(map[string]*Client)myMap[c.User.UserId] = c                //userIdc.Hub.ClientsMap[c.User.RoomId] = myMap //roomIdflag = true}return
}// DeleteFromHub 写操作 --逻辑删除 将传入的参数c从hub连接池中删除
func (h *HupCenter) DeleteFromHub(c *Client) {h.mutex.Lock()defer h.mutex.Unlock()if c.User.RoomId == "" {return}if value, ok1 := c.Hub.ClientsMap[c.User.RoomId]; ok1 {if _, ok2 := value[c.User.UserId]; ok2 {delete(value, c.User.UserId)}}if len(c.Hub.ClientsMap[c.User.RoomId]) == 0 {delete(c.Hub.ClientsMap, c.User.RoomId)}
}// QueryOtherUser 读操作 -- 根据当前用户寻找另一位用户，返回user对象
func (h *HupCenter) QueryOtherUser(c *Client) *Client {if roomMap, ok := h.ClientsMap[c.User.RoomId]; ok { //roomfor userId, user := range roomMap {if userId != c.User.UserId {return user}}}return nil
}

3.管道解决方案

使用锁是能够基本解决问题的，但是对于读写较为频繁的场景，读写锁可能会成为性能瓶颈，再加上自己对管道的运用不是很熟练，就开始思考如何使用channel去解决这一个并发的问题，代码如下：

type HupCenter struct {ClientsMap map[string]map[string]*Client `json:"-"` //第一个string-roomId 第二个string-userIdRegister   chan *ClientUnRegister chan *Client
}// NewHub 初始化一个hub
func NewHub() *HupCenter {return &HupCenter{ClientsMap: make(map[string]map[string]*Client),Register:   make(chan *Client, 1),UnRegister: make(chan *Client, 1),}
}// Run 用户向hub中的逻辑注册、删除、心跳检测全方法,在代码执行后，开始协程去执行Run方法
func (h *HupCenter) Run() {checkTicker := time.NewTicker(time.Duration(pkg.HeartCheckSecond) * time.Second)defer checkTicker.Stop()for {select {case client := <-h.Register://先查询是否存在此一个roomId keyif myMap, ok := client.Hub.ClientsMap[client.User.RoomId]; ok { //有,加入房间//检测人数if len(myMap) == 1 {myMap[client.User.UserId] = client}} else { //没有,创建房间myMap := make(map[string]*Client)myMap[client.User.UserId] = client                //userIdclient.Hub.ClientsMap[client.User.RoomId] = myMap //roomId}fmt.Println("有人加入房间:", client.Hub.ClientsMap)case client := <-h.UnRegister:client.User.Close()if value, ok1 := client.Hub.ClientsMap[client.User.RoomId]; ok1 {if _, ok2 := value[client.User.UserId]; ok2 {delete(value, client.User.UserId)}}if len(client.Hub.ClientsMap[client.User.RoomId]) == 0 {delete(client.Hub.ClientsMap, client.User.RoomId)}case <-checkTicker.C:for _, roomMap := range h.ClientsMap {for _, client := range roomMap {if client.User.HealthCheck.Before(time.Now()) {h.UnRegister <- client}}}fmt.Println(time.Now().Format(time.DateTime), h.ClientsMap)}}
}// QueryOtherUser 读操作 -- 根据当前用户寻找另一位用户，返回user对象
func (h *HupCenter) QueryOtherUser(c *Client) *Client {if roomMap, ok := h.ClientsMap[c.User.RoomId]; ok { //roomfor userId, user := range roomMap {if userId != c.User.UserId {return user}}}return nil
}

在代码中，我们在结构体中定义了两个管道，一个管道接收注册的客户端对象（原JoinHub方法），另一个管道接收注销的客户端对象（原DeleteFormHub方法);

在Run方法中，我们创建了一个10秒的ticker对象，来进行客户端连接的心跳检测。之后使用for循环来执行select来监听多个管道，并执行对应的分支操作。select会随机挑选一个可执行的case语句，如果没有可执行的case,则进行等待。在本代码中如果没有注册、注销的操作，会每隔10秒进行一次心跳检测，并打印当前存活的客户端对象集合。