当前位置：首页 > web >正文

Go 即时通讯系统：客户端与服务端 WebSocket 通信交互

web 2025/7/15 8:45:57

客户端和服务端的交互

客户端与服务端建立连接

客户端：客户端通过浏览器或者其他应用程序发起一个 HTTP 请求到服务端的 /socket.io 路径。在请求中会携带用户的 UUID 作为参数（通过 c.Query("user") 获取）。

// router/socket.go
func RunSocket(c *gin.Context) {user := c.Query("user") // 获取用户标识if user == "" {return // 无用户标识则拒绝连接}log.Info("newUser", log.String("newUser", user))// 升级 HTTP 连接为 WebSocket 连接ws, err := upGrader.Upgrade(c.Writer, c.Request, nil)if err != nil {return}// 创建客户端对象client := &server.Client{Name: user,Conn: ws,Send: make(chan []byte),}// 注册客户端到服务器server.MyServer.Register <- client// 启动读写协程go client.Read()go client.Write()
}

服务端：服务端接收到请求后，使用 websocket.Upgrader 将 HTTP 连接升级为 WebSocket 连接。然后创建一个 Client 实例，并将其发送到 Server 的 Register 通道。

客户端注册到服务端

// internal/server/server.go
func (s *Server) Start() {for {select {// 处理客户端注册case conn := <-s.Register:log.Info("login", log.String("login", conn.Name))s.Clients[conn.Name] = connmsg := &protocol.Message{From:    "System",To:      conn.Name,Content: "welcome!",}protoMsg, _ := proto.Marshal(msg) // 序列化为字节切片conn.Send <- protoMsg}}
}

客户端：无特定操作，等待服务端响应。
服务端：服务端的 Start 方法会监听 Register 通道，当有新的客户端注册时，将客户端信息保存到 Clients 映射中，并向客户端发送欢迎消息。

客户端发送消息到服务端

客户端：客户端通过 Client 结构体的 Read 方法从 WebSocket 连接读取消息。如果是心跳消息，客户端会发送 Pong 响应；否则，根据配置将消息发送到 Kafka 或者服务端的 Broadcast 通道。

// internal/server/client.go
func (c *Client) Read() {// 消息读取for {c.Conn.PongHandler()_, message, err := c.Conn.ReadMessage()if err != nil {log.Error("client read message error", log.Err(err))MyServer.Ungister <- cc.Conn.Close()break}msg := &protocol.Message{}proto.Unmarshal(message, msg)// 处理心跳消息if msg.Type == constant.HEAT_BEAT {pong := &protocol.Message{Content: constant.PONG,Type:    constant.HEAT_BEAT,}pongBytes, err2 := proto.Marshal(pong)if err2 != nil {log.Error("client marshal message error", log.Err(err))}c.Conn.WriteMessage(websocket.BinaryMessage, pongBytes)} else {MyServer.Broadcast <- message}}
}

服务端：服务端的 Start 方法会监听 Broadcast 通道，当接收到消息时，根据消息的类型（单聊、群聊）将消息转发给相应的客户端。

// internal/server/server.go
func (s *Server) Start() {for {select {// 处理消息广播case message := <-s.Broadcast:msg := &protocol.Message{}proto.Unmarshal(message, msg)if msg.To != "" {// 有指定接收者的消息处理if msg.ContentType >= constant.TEXT && msg.ContentType <= constant.VIDEO {// 一般消息，比如文本消息，视频文件消息等_, exists := s.Clients[msg.From]if exists { // 检查发送者是否在连接列表中saveMessage(msg)}if msg.ContentType == constant.MESSAGE_TYPE_USER {// 单人消息处理s.sendUserMessage(msg)} else if msg.ContentType == constant.MESSAGE_TYPE_GROUP {// 多人消息处理s.sendGroupMessage(msg)} else {// 语音电话，视频电话等，仅支持单人聊天，不支持群聊// 不保存文件，直接进行转发client, ok := s.Clients[msg.To]if ok {client.Send <- message}}} else {// 无指定接收者的广播消息处理for id, conn := range s.Clients {log.Info("allUser", log.String("allUser", id))select {case conn.Send <- message: // 发送消息给客户端，成功继续处理default: // 失败关闭客户端close(conn.Send)delete(s.Clients, conn.Name)}}}}}}
}

服务端发送消息到客户端

服务端：服务端将消息发送到客户端的 Send 通道。

// internal/server/server.go
client.Send <- msgByteclient.Send <- message

客户端：客户端的 Write 方法会监听 Send 通道，当接收到消息时，将消息通过 WebSocket 连接发送给客户端。

// internal/server/client.go
func (c *Client) Write() {defer func() {c.Conn.Close()}()for message := range c.Send {c.Conn.WriteMessage(websocket.BinaryMessage, message)}
}

客户端断开连接

客户端：无特定操作，等待服务端响应。
服务端：服务端的 Start 方法会监听 Ungister 通道，当有客户端断开连接时，将客户端信息从 Clients 映射中删除，并关闭客户端的 Send 通道。

// internal/server/server.go
func (s *Server) Start() {for {// 处理客户端注销case conn := <-s.Ungister:log.Info("loginout", log.String("loginout", conn.Name))if _, ok := s.Clients[conn.Name]; ok {close(conn.Send)delete(s.Clients, conn.Name)}}
}