fix

server/websocket/internal/logic/commonnotifylogic.go

@@ -5,7 +5,6 @@ import (
 	"fusenapi/constants"
 	"fusenapi/utils/auth"
 	"fusenapi/utils/basic"
-	"sync"
 	"time"
 
 	"fusenapi/server/websocket/internal/svc"
@@ -37,7 +36,7 @@ type commonConnectionNotFoundDataCacheChanItem struct {
 }
 
 // 放入缓冲队列
-func (l *CommonNotifyLogic) pushCommonNotifyCache(data commonConnectionNotFoundDataCacheChanItem) {
+func pushCommonNotifyCache(data commonConnectionNotFoundDataCacheChanItem) {
 	select {
 	case commonConnectionNotFoundDataCacheChan <- data:
 		return
@@ -47,20 +46,26 @@ func (l *CommonNotifyLogic) pushCommonNotifyCache(data commonConnectionNotFoundD
 }
 
 // 取出元素
-func (l *CommonNotifyLogic) popCommonNotifyCache() (data commonConnectionNotFoundDataCacheChanItem) {
+func popCommonNotifyCache() (data commonConnectionNotFoundDataCacheChanItem) {
 	return <-commonConnectionNotFoundDataCacheChan
 }
 
-// 保证处理消息就一个循环在执行
-var consumeCommonCacheData sync.Once
-
 // 消费公共通知未处理的消息(目前是轮巡方式，待优化)
-func (l *CommonNotifyLogic) consumeCommonCacheData() {
-	//单例
-	consumeCommonCacheData.Do(func() {
+func ConsumeCommonCacheData(ctx context.Context) {
+	defer func() {
+		if err := recover(); err != nil {
+			logx.Error("consumeCommonCacheData panic :", err)
+		}
+	}()
+	go func() {
+		select {
+		case <-ctx.Done():
+			panic("consumeCommonCacheData ctx deadline")
+		}
+	}()
 	for {
 		time.Sleep(time.Millisecond * 200)
-			info := l.popCommonNotifyCache()
+		info := popCommonNotifyCache()
 		//查询websocket连接
 		value, ok := mapConnPool.Load(info.data.Wid)
 		//没有连接
@@ -68,7 +73,7 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
 			info.retryTimes--
 			//大于0，则放回队列
 			if info.retryTimes > 0 {
-					l.pushCommonNotifyCache(info)
+				pushCommonNotifyCache(info)
 				continue
 			}
 			//否则直接丢弃消息
@@ -83,7 +88,6 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
 		//发送
 		ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data))
 	}
-	})
 }
 
 // 处理进入前逻辑w,r
@@ -91,17 +95,20 @@ func (l *CommonNotifyLogic) consumeCommonCacheData() {
 // }
 
 func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) {
-	//websocket连接id不能为空
+	searchConnectType := "uniqueId"
 	if req.Wid == "" {
+		if !userinfo.IsUser() && !userinfo.IsGuest() {
 			return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "websocket connect id is empty")
 		}
-	//触发消费公共未处理的消息（该方法是单例）
-	go l.consumeCommonCacheData()
+		searchConnectType = "userInfo"
+	}
+	switch searchConnectType {
+	case "uniqueId": //直接通过唯一标识发消息
 		//查询websocket连接
 		value, ok := mapConnPool.Load(req.Wid)
 		if !ok {
 			//没找到连接就放到公共缓冲队列
-		go l.pushCommonNotifyCache(commonConnectionNotFoundDataCacheChanItem{
+			pushCommonNotifyCache(commonConnectionNotFoundDataCacheChanItem{
 				retryTimes: 20, //重试20次
 				data: types.CommonNotifyReq{
 					Wid:  req.Wid,
@@ -117,6 +124,9 @@ func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *a
 			return resp.SetStatusWithMessage(basic.CodeServiceErr, "断言连接错误")
 		}
 		ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, req.Data))
+	case "userInfo": //通过用户信息找连接发送
+		sendToOutChanByUserIndex(userinfo.UserId, userinfo.GuestId, (&wsConnectItem{}).respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, req.Data))
+	}
 	return resp.SetStatusWithMessage(basic.CodeOK, "success")
 }
 

server/websocket/internal/logic/datatransferlogic.go

@@ -148,8 +148,6 @@ func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request)
 		conn.Close()
 		return
 	}
-	//消费用户索引控制chan的数据
-	go consumeUserPoolData()
 	//循环读客户端信息
 	go ws.acceptBrowserMessage()
 	//消费出口数据并发送浏览器端
@@ -257,16 +255,19 @@ func sendToOutChanByUserIndex(userId, guestId int64, message []byte) {
 	}
 }
 
-// 消费用户索引池中的任务(单例)
-var consumeUserPoolDataOnce sync.Once
-
-func consumeUserPoolData() {
+// 消费用户索引创建/删除/发送消息中的任务数据
+func ConsumeUserPoolData(ctx context.Context) {
 	defer func() {
 		if err := recover(); err != nil {
 			logx.Error("consumeUserPoolData panic:", err)
 		}
 	}()
-	consumeUserPoolDataOnce.Do(func() {
+	go func() {
+		select {
+		case <-ctx.Done():
+			panic("ConsumeUserPoolData ctx deadline")
+		}
+	}()
 	for {
 		select {
 		case data := <-mapUserConnPoolCtlChan:
@@ -308,7 +309,6 @@ func consumeUserPoolData() {
 			}
 		}
 	}
-	})
 }
 
 // 获取mapUserConnPool唯一id

server/websocket/websocket.go

@@ -1,8 +1,10 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
+	"fusenapi/server/websocket/internal/logic"
 	"net/http"
 
 	"fusenapi/utils/auth"
@@ -28,6 +30,13 @@ func main() {
 
 	ctx := svc.NewServiceContext(c)
 	handler.RegisterHandlers(server, ctx)
+	ctx1 := context.Background()
+	ctx1, cancel := context.WithCancel(ctx1)
+	defer cancel()
+	//消费公共通知队列的数据
+	go logic.ConsumeCommonCacheData(ctx1)
+	//消费用户索引创建/删除/发送消息中的任务数据
+	go logic.ConsumeUserPoolData(ctx1)
 	fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
 	server.Start()
 }