Merge branch 'dev' of http://gitea-inner.fontree.cn/fiee/fonchain-fiee into dev

pkg/common/ws/chatRoom.go

@@ -118,6 +118,8 @@ func (o *ChatRoom) Run() {
 						}
 					}
 				}
+				// 将自己加入会话
+				o.Session[newClient.SessionId] = append(o.Session[newClient.SessionId], newClient)
 			} else {
 				//普通用户添加会话的逻辑
 				_, ok := o.Session[newClient.SessionId]
@@ -136,10 +138,10 @@ func (o *ChatRoom) Run() {
 						}
 					}
 				}
-				o.clientsRwLocker.Unlock()
-				//再把自己的客户端加入会话
+				// 将自己加入会话
 				o.Session[newClient.SessionId] = append(o.Session[newClient.SessionId], newClient)
 			}
+			o.clientsRwLocker.Unlock() // 统一在最后解锁
 			o.pushEvent(EventUserJoin, EventProgressAfter, nil, newClient)
 		//注销事件
 		case client := <-o.UnRegister:
@@ -282,42 +284,49 @@ func (o *ChatRoom) GetUserIdInSession(sessionId string, withoutUserId ...int64)
 //		o.unRegister <- c
 //	}
 func (o *ChatRoom) Broadcast(message []byte, userIds ...int64) {
+	fmt.Println("Broadcast -------1")
 	// 如果userIds为空则群发,否则找到这个用户的ws对象
+	var clientsToSend []*Client
+
 	if userIds == nil {
 		for _, userClients := range o.clients {
 			for _, cli := range userClients {
-				if cli == nil {
-					o.UnRegister <- cli
-					continue
+				if cli != nil {
+					clientsToSend = append(clientsToSend, cli)
 				}
-				go func() {
-					err := cli.Conn.WriteMessage(websocket.TextMessage, message)
-					if err != nil {
-						o.UnRegister <- cli
-					}
-				}()
 			}
 		}
 	} else {
 		for _, userId := range userIds {
 			userClients, ok := o.clients[userId]
-			if ok == false {
-				return
+			if !ok {
+				continue
 			}
 			for _, cli := range userClients {
-				if cli == nil {
-					o.UnRegister <- cli
-					continue
+				if cli != nil {
+					clientsToSend = append(clientsToSend, cli)
 				}
-				go func() {
-					err := cli.Conn.WriteMessage(websocket.TextMessage, message)
-					if err != nil {
-						o.UnRegister <- cli
-					}
-				}()
 			}
 		}
 	}
+
+	// 使用有限的goroutine池发送消息
+	fmt.Println("Broadcast -------2")
+	var wg sync.WaitGroup
+	for _, cli := range clientsToSend {
+		wg.Add(1)
+		go func(client *Client) {
+			defer wg.Done()
+
+			err := client.Conn.SetWriteDeadline(time.Now().Add(writeWait))
+			err = client.Conn.WriteMessage(websocket.TextMessage, message)
+			if err != nil {
+				o.UnRegister <- client
+			}
+		}(cli)
+	}
+	wg.Wait()
+	fmt.Println("Broadcast -------3 end")
 }
 
 // RegisterEventListener 注册聊天室事件监听者

pkg/common/ws/client.go

@@ -94,15 +94,21 @@ func (c *Client) WriteWait() {
 	for {
 		select {
 		case msg, ok := <-c.Send:
-			fmt.Printf("发送消息:%+v\n", string(msg))
-			c.Conn.WriteMessage(websocket.TextMessage, msg)
-			fmt.Println("发送消息结束")
 			if !ok {
 				// 聊天室关闭了管道
 				c.Conn.WriteControl(websocket.CloseMessage, []byte{}, time.Now().Add(writeWait))
 				return
 			}
+			// 设置写入超时
+			err := c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
+			fmt.Printf("设置写超时 err check:%v\n", err)
+			fmt.Printf("发送消息:%+v\n", string(msg))
+			err = c.Conn.WriteMessage(websocket.TextMessage, msg)
+			fmt.Printf("发送消息结束 err check:%v\n", err)
 		case <-ticker.C:
+			fmt.Println("ping websocket client")
+			err := c.Conn.SetWriteDeadline(time.Now().Add(writeWait))
+			fmt.Printf("ping 设置写超时 err check:%v\n", err)
 			if err := c.Conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(pongWait)); err != nil {
 				return
 			}

pkg/service/asChat/robot/readme.md

@@ -0,0 +1 @@
+# robot 聊天机器人

pkg/service/asChat/robot/robot.go

@@ -197,6 +197,8 @@ func (r *Robot) RegisterDelayTask(task RobotTask) {
 	}
 	r.DelayTask = append(r.DelayTask, task)
 }
+
+// 重载回复规则
 func (r *Robot) ReloadRules(ctx context.Context) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()