gin框架中使用websocket发送消息及群聊

一、依赖包的安装

• 1、在go语言中常见的websocket包有以下两个

  • github地址,优先选择别人封装好的
  • 官方包

• 2、选择set集合包，链接地址

• 3、在gin框架中使用，链接地址

二、在gin中使用websocket

• 1、接入鉴权

  websocket也可以和普通api接口一样的做一个接口鉴权(token机制),如果验证通过可以继续往下走,没有验证不能往下走

  func Chat(ctx *gin.Context) {
  	var upGrader = websocket.Upgrader{
  		CheckOrigin: func(r *http.Request) bool {
  			// 根据鉴权的方式来处理,如果不想鉴权的就直接返回true,如果需要鉴权就要根据判断来返回true，或者false
  			return true
  		},
  	}
  }
  1
  2
  3
  4
  5
  6
  7
  8

• 2、实现鉴权处理

  func Chat(ctx *gin.Context) {
    // 根据url地址上获取当前用户id和token
  	userId := ctx.DefaultQuery("userId", "")
  	token := ctx.DefaultQuery("token", "")
  	userIdInt, _ := strconv.ParseInt(userId, 10, 64)
  	isValied := checkToken(userIdInt, token)
  	var upGrader = websocket.Upgrader{
  		CheckOrigin: func(r *http.Request) bool {
  			// 根据判断token的方法来鉴权,如果没token就返回false
  			return isValied
  		},
  	}
  }
  
  // 内部使用判断token合法
  func checkToken(userId int64, token string) bool {
  	...
  	return true
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19

• 3、将普通的get请求升级为websocket请求

  ...
  //升级get请求为webSocket协议
  conn, err := upGrader.Upgrade(ctx.Writer, ctx.Request, nil)
  if err != nil {
    fmt.Println("websocket连接错误")
    return
  }
  1
  2
  3
  4
  5
  6
  7

• 4、conn连接句柄的维护

  // Node 当前用户节点 userId和Node的映射关系
  type Node struct {
  	Conn      *websocket.Conn
  	DataQueue chan []byte
    // 群组的消息分发
  	GroupSet  set.Interface
  }
  
  // 映射关系表
  var clientMap map[int64]*Node = make(map[int64]*Node, 0)
  
  // 读写锁
  var rwLocker sync.RWMutex
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

• 5、每次创建连接后将映射关系进行绑定

  func Chat(ctx *gin.Context) {
  	userId := ctx.DefaultQuery("userId", "")
  	token := ctx.DefaultQuery("token", "")
  	userIdInt, _ := strconv.ParseInt(userId, 10, 64)
  	fmt.Println(token, userId, "=======")
  	isValied := checkToken(userIdInt, token)
  	var upGrader = websocket.Upgrader{
  		CheckOrigin: func(r *http.Request) bool {
  			// 根据判断token的方法来鉴权,如果没token就返回false
  			return isValied
  		},
  	}
  	//升级get请求为webSocket协议
  	conn, err := upGrader.Upgrade(ctx.Writer, ctx.Request, nil)
  	if err != nil {
  		fmt.Println("websocket连接错误")
  		return
  	}
  	// 绑定到当前节点
  	node := &Node{
  		Conn:      conn,
  		DataQueue: make(chan []byte, 50),
  		GroupSet:  set.New(set.ThreadSafe),
  	}
  	// 映射关系的绑定
  	rwLocker.Lock()
  	clientMap[userIdInt] = node
  	rwLocker.Unlock()
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29

• 6、创建一个发送消息到管道中

  // 将数据推送到管道中
  func sendMsg(userId int64, message []byte) {
  	rwLocker.RLock()
  	node, isOk := clientMap[userId]
  	rwLocker.RUnlock()
  	if isOk {
  		node.DataQueue <- message
  	}
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9

• 7、创建一个方法从管道中获取数据发送给前端

  // 从管道中获取数据发送出去
  func senProc(node *Node) {
  	for {
  		select {
  		case data := <-node.DataQueue:
  			err := node.Conn.WriteMessage(websocket.TextMessage, data)
  			if err != nil {
  				fmt.Println("发送消息失败")
  				return
  			}
  		}
  	}
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

• 8、在Chat方法中使用

  func Chat(ctx *gin.Context) {
    ...
    // 连接成功就给当前用户发送一个hello word
  	sendMsg(userIdInt, []byte("hello word"))
  	// 发送数据给客户端
  	go senProc(node)
  }
  1
  2
  3
  4
  5
  6
  7

• 9、接收客户端消息转发给另外一个用户

  // 接收客户端数据
  func recvProc(node *Node) {
  	for {
  		_, data, err := node.Conn.ReadMessage()
  		if err != nil {
  			fmt.Println("接收数据失败", err)
  			return
  		}
  		// 将数据处理转发给对应的人
  		dispatch(data)
  	}
  }
  
  // 分发数据
  func dispatch(data []byte) {
  	type Message struct {
  		UserId int64  `json:"userId"`
  		Msg    string `json:"msg"`
  	}
  	fmt.Println("接收到的数据", string(data))
  	// 解析出数据
  	message := Message{}
  	err := json.Unmarshal(data, &message)
  	if err != nil {
  		fmt.Println("解析数据失败:", err.Error())
  		return
  	}
  	fmt.Println("解析的数据为:", message)
  	// 发送数据
  	sendMsg(message.UserId, data)
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31

• 10、使用接收客户端数据的方法

  func Chat(ctx *gin.Context) {
    ...
    // 连接成功就给当前用户发送一个hello word
  	sendMsg(userIdInt, []byte("hello word"))
  	// 发送数据给客户端
  	go senProc(node)
    // 接收客户端的数据
    go recvProc(node)
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9

• 11、定义一个对外的方法(比如在别的接口中要发送数据到websocket中)

  func SendMessage(userId int64, message interface{}) {
  	str, _ := json.Marshal(message)
  	sendMsg(userId, str)
  }
  1
  2
  3
  4

三、消息群聊的使用

常见的场景有群聊，一个后台用户要给自己的顾客推送促销消息，这里就举例用后台给顾客推送促销消息

• 1、根据当前连接的用户id来加入对应的群聊

  // addGroupByAccountId 加入群聊
  func AddGroupByAccountId(userId, groupId int64) {
  	fmt.Println(userId, "加入群聊")
  	rwLocker.Lock()
  	node, isOk := FrontClientMap[userId]
  	if isOk {
  		node.GroupSet.Add(groupId)
  		fmt.Println("加入群聊成功")
  	}
  	rwLocker.Unlock()
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

• 2、在连接的时候加入群聊

  func Chat(ctx *gin.Context) {
  	...
  	// 根据当前用户的id来加入群组
  	AddGroupByAccountId(userIdInt)
    ...
  }
  1
  2
  3
  4
  5
  6

• 3、推送消息，循环连接的Map判断如果当前用户在这个群聊里面就发送数据到管道中

  // 2.websocket推送消息到h5群端
  for _, v := range frontWs.FrontClientMap {
    data := gin.H{
      "messageType": int64(messageDto.MessageType),
      "title":       messageDto.Title,
      "content":     messageDto.Content,
    }
    if v.GroupSet.Has(int64(accountId)) {
      msg, _ := json.Marshal(data)
      v.DataQueue <- msg
    }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12

• 4、处理群聊过程中临时加入群聊的，直接在加入的时候调用加入群组的方法就可以