// 房间管理模块 -- 负责创建、管理聊天房间,处理客户端的加入、离开和消息转发 package wsClient import ( "encoding/base64" "encoding/json" "fmt" "math/rand" "net/http" "net/url" "strings" "sync" "chatapp/external" "github.com/gorilla/websocket" "github.com/rs/zerolog/log" ) // 全局房间 map,存储所有创建的房间 // key: 房间名称,value: 房间对象指针 var rooms = make(map[string]*Room) // Room 表示一个聊天房间 type Room struct { // holds all current clientConns in the room // 保存房间中所有当前活跃的客户端连接 clientConns map[*Client]bool // join is a channel for all clients wishing to join the room // 用于接收希望加入房间的客户端。新客户端通过这个 channel 请求加入 join chan *Client // leave is a channel for all clients wishing to leave the room // 用于接收希望离开房间的客户端。断开连接的客户端通过这个 channel 通知离开 leave chan *Client // forward is a channel that holds incoming messages that should be forwarded to the other clients. // 用于接收需要转发给其他客户端的消息 // 所有向此房间发送的消息都会通过这个 channel 进行分发 forward chan *message // name 房间名称,用作Redis存储的键 name string } // message 包含要转发的消息内容和发送者信息 type message struct { content []byte // 消息内容 sender *Client // 发送者 -- // 用处:1.用于在转发时排除发送者自己 } // RoomParams 表示从客户端 WebSocket子协议中解析出的房间参数 type RoomParams struct { RoomName string `json:"roomName,omitempty"` UserId int32 `json:"u_id"` // 用户ID: 只是表示买家客户端是 userId CustomId int32 `json:"c_id"` ShopId int32 `json:"s_id"` Username string `json:"u_name"` Platform string `json:"platform"` Type string `json:"type,omitempty"` // 客户端类型:shop-门店端, customer-客户端 StaffId int32 `json:"staff_id"` // 只是表示门店客户端的员工Id } // 全局互斥锁,用于保护 rooms 字典的并发访问,确保多个 goroutine 同时访问 rooms 时的线程安全 var mu sync.Mutex // newRoom 创建并返回一个新的房间实例 // 初始化所有必要的 channel 和 map func newRoom(name string) *Room { return &Room{ clientConns: make(map[*Client]bool), join: make(chan *Client), leave: make(chan *Client), forward: make(chan *message), // 现在传输 message 结构体 name: name, // 设置房间名称 } } // GetRoomAndRun 获取指定名称的房间,如果不存在则创建新房间并启动 // 这是线程安全的函数,使用互斥锁保护全局房间字典 func GetRoomAndRun(name string) *Room { mu.Lock() defer mu.Unlock() if r, ok := rooms[name]; ok { return r } r := newRoom(name) // 传入房间名称 rooms[name] = r // 启动房间的消息处理循环(在新的 goroutine 中) go r.run() return r } // run 是房间的核心消息处理循环 // 在独立的 goroutine 中运行,使用 select 语句处理三种类型的事件 func (r *Room) run() { for { select { // 处理客户端加入事件 case client := <-r.join: // 检查channel是否已关闭(收到nil值) if client == nil { return // channel已关闭,退出goroutine } // 将新客户端添加到房间的客户端集合中 r.clientConns[client] = true // 使用 true 作为值,实际上只使用 key(client 指针) // log.Printf("客户端 %v 加入房间 %v", client.name, r.name) log.Info().Str("clientName", client.name).Str("roomName", r.name).Msg("客户端加入房间") // 清空未读消息数 go ClearUnreadCount(client) // 处理客户端离开事件 case client := <-r.leave: // 检查channel是否已关闭(收到nil值) if client == nil { return // channel已关闭,退出goroutine } // 从房间的客户端集合中删除客户端 delete(r.clientConns, client) // 关闭客户端的接收 channel,通知客户端的 write() 方法退出 close(client.receive) log.Info().Str("clientName", client.name).Str("roomName", r.name).Msg("客户端离开房间") if len(r.clientConns) == 0 { log.Info().Str("roomName", r.name).Msg("房间没有客户端,关闭房间") close(r.forward) close(r.join) close(r.leave) delete(rooms, r.name) return // 退出run() goroutine,避免继续从已关闭的channel读取数据 } // 处理消息转发事件 case msg := <-r.forward: // 检查channel是否已关闭(收到nil值) if msg == nil { return // channel已关闭,退出goroutine } // 打印消息内容与发送者信息 log.Debug().Str("sender", msg.sender.name).Str("msg", string(msg.content)).Msg("Received message") // 直接存储原始消息内容到Redis,使用房间名作为键 go StoreRawMessage(r.name, msg.content) // 异步存储,不阻塞消息转发 // 创建最近联系人列表 或 更新最近联系人列表 go UpdateRecentContacts(msg.sender) online := false // 将消息发送给房间中的所有客户端,但排除发送者自己 for client := range r.clientConns { if client != msg.sender { // 高效的指针比较,排除发送者 // && client.userInfo.Type != msg.sender.userInfo.Type -- 排除同一类型的客户端 // 将消息发送到每个客户端的接收 channel client.receive <- msg.content // 只发送消息内容,不包含发送者信息 online = true } } // 本消息转发,目的是让客户与门店互相发消息,当有另一方不在线时,要累计未读消息数 if !online { go IncrementUnreadCount(msg.sender, 1) receiver, shopId, userId, count := GeneHaveSendInfo(msg.sender, 1) // 向 PHP 服务发送未读消息通知 go external.RequestToPHP(receiver, shopId, userId, count) } } } } // 查询当前房间的客户端数量 func (r *Room)clientCount() int { return len(r.clientConns) } // 统计所有房间里的客户端总数量 func GetTotalClientCount() int { totalCount := 0 for _, room := range rooms { totalCount += len(room.clientConns) } return totalCount } // 用 Client.name 来查找房间 func GetRoomByClientName(clientName string) *Room { for _, room := range rooms { for client := range room.clientConns { if client.name == clientName { return room } } } return nil } // 用 Client.name 查询用户是否在线 func IsClientOnline(clientName string) bool { room := GetRoomByClientName(clientName) return room != nil } // ======================================== webSocket服务相关 ======================================== // WebSocket 相关常量定义 const ( // WebSocket 连接的读写缓冲区大小(字节) socketBufferSize = 1024 // 每个客户端消息接收 channel 的缓冲区大小 messageBufferSize = 256 ) // WebSocket 升级器,用于将 HTTP 连接升级为 WebSocket 连接 // 设置读写缓冲区大小以优化性能 var upgrader = &websocket.Upgrader{ ReadBufferSize: socketBufferSize, // 读缓冲区大小 WriteBufferSize: socketBufferSize, // 写缓冲区大小 CheckOrigin: func(r *http.Request) bool { return true // 允许跨域连接 }, } // parseWebSocketProtocols 解析WebSocket子协议中的参数 // 客户端通过子协议传递多个Base64编码的JSON参数片段 func ParseWebSocketProtocols(protocols []string) (*RoomParams, error) { //log.Println("-------- ParseWebSocketProtocols ---------") // 查找包含房间参数的子协议 for _, protocol := range protocols { //log.Printf("Parsing protocol: %s", protocol) if strings.HasPrefix(protocol, "chat,") { // 初始化参数结构体 params := RoomParams{} // 分割协议字符串,获取各个参数部分 parts := strings.Split(protocol, ",") if len(parts) < 2 { log.Warn().Str("protocol", protocol).Msg("协议格式错误,参数不足") continue } // 遍历所有参数部分(跳过第一个"chat") for i := 1; i < len(parts); i++ { part := strings.TrimSpace(parts[i]) if err := parseProtocolPart(part, ¶ms); err != nil { // log.Printf("解析协议部分失败 %s: %v", part, err) log.Error().Err(err).Str("part", part).Msg("解析协议部分失败") continue } } // log.Printf("成功解析WebSocket协议参数: %+v\n", params) log.Info().Any("params", params).Msg("成功解析WebSocket协议参数") return ¶ms, nil } else { log.Debug().Str("protocol", protocol).Msg("跳过非chat协议") } } return nil, fmt.Errorf("未找到有效的房间参数协议") } // parseProtocolPart 解析单个协议参数部分 func parseProtocolPart(part string, params *RoomParams) error { // 检查参数格式:p0-..., p1-..., p2-... if len(part) < 3 || !strings.Contains(part, "-") { return fmt.Errorf("参数格式错误: %s", part) } // 分割前缀和base64内容 sepIndex := strings.Index(part, "-") if sepIndex == -1 { return fmt.Errorf("缺少分隔符: %s", part) } prefix := part[:sepIndex] encodedData := part[sepIndex+1:] // Base64解码 - 按优先级尝试不同的编码方式 decodedBytes, err := base64.RawStdEncoding.DecodeString(encodedData) if err != nil { // 如果RawStdEncoding失败,尝试RawURLEncoding(URL安全的base64) decodedBytes, err = base64.RawURLEncoding.DecodeString(encodedData) if err != nil { // 如果RawURLEncoding也失败,尝试标准的StdEncoding(带填充) decodedBytes, err = base64.StdEncoding.DecodeString(encodedData) if err != nil { // 最后尝试URLEncoding decodedBytes, err = base64.URLEncoding.DecodeString(encodedData) if err != nil { return fmt.Errorf("Base64解码失败: %v", err) } } } } // URL解码 decodedStr, err := url.QueryUnescape(string(decodedBytes)) if err != nil { return fmt.Errorf("URL解码失败: %v", err) } // 根据前缀类型解析不同的参数 switch prefix { case "p0": // 解析用户ID和自定义ID var p0Data struct { UserId int32 `json:"u_id"` CustomId int32 `json:"c_id"` } if err := json.Unmarshal([]byte(decodedStr), &p0Data); err != nil { return fmt.Errorf("p0参数JSON解析失败: %v", err) } params.UserId = p0Data.UserId params.CustomId = p0Data.CustomId case "p1": // 解析店铺ID和用户名 var p1Data struct { ShopId int32 `json:"s_id"` Username string `json:"u_name"` } if err := json.Unmarshal([]byte(decodedStr), &p1Data); err != nil { return fmt.Errorf("p1参数JSON解析失败: %v", err) } params.ShopId = p1Data.ShopId params.Username = p1Data.Username case "p2": // 解析平台和类型 var p2Data struct { Platform string `json:"platform"` Type string `json:"type"` } if err := json.Unmarshal([]byte(decodedStr), &p2Data); err != nil { return fmt.Errorf("p2参数JSON解析失败: %v", err) } params.Platform = p2Data.Platform params.Type = p2Data.Type case "p3": // 解析平台和类型 var p3Data struct { StaffId int32 `json:"staff_id"` } if err := json.Unmarshal([]byte(decodedStr), &p3Data); err != nil { return fmt.Errorf("p2参数JSON解析失败: %v", err) } params.StaffId = p3Data.StaffId // 当有 p4 参数时,解析 p4 参数 default: log.Warn().Str("prefix", prefix).Msg("未知的参数前缀") } return nil } func (r *Room) HttpServe(w http.ResponseWriter, req *http.Request,roomParams *RoomParams) { // 将 HTTP 连接升级为 WebSocket 连接,指定接受的子协议 responseHeader := http.Header{} // 这行代码的作用是在 WebSocket握手响应中添加 Sec-WebSocket-Protocol头,值为"chat"。这表示服务器接受并选择了"chat"子协议作为通信协议。当客户端在请求中提供多个子协议选项时,服务器需要在响应中指定它选择使用哪一个。这行代码确认服务器将使用基础的"chat"协议与客户端通信,这样客户端就知道应该使用哪种协议格式来解释后续的WebSocket消息。 responseHeader.Add("Sec-WebSocket-Protocol", "chat") // 响应基础的 chat 协议 socket, err := upgrader.Upgrade(w, req, responseHeader) if err != nil { log.Error().Err(err).Msg("Upgrade error") return } // ----------------------------------- 生成客户端对象 ---------------------------------- // 创建新的客户端对象,使用解析出的参数 client := &Client{ name: clientName(roomParams), socket: socket, // WebSocket 连接 receive: make(chan []byte, messageBufferSize), // 接收消息的缓冲 channel room: r, // 使用当前房间实例 userInfo: UserInfo{ UserId: roomParams.UserId, Username: roomParams.Username, CustomId: roomParams.CustomId, ShopId: roomParams.ShopId, Platform: roomParams.Platform, Type: roomParams.Type, StaffId: roomParams.StaffId, }, } // 如果用户名为空,生成一个默认用户名 if client.name == "" { client.name = fmt.Sprintf("user_%d", rand.Intn(9000000)) } // 将客户端加入当前房间 r.join <- client defer func() { r.leave <- client }() // 启动客户端的写消息 goroutine go client.write() // 在当前 goroutine 中处理客户端的读消息(阻塞直到连接断开) client.read() }