package core import ( "JsRpc/config" "JsRpc/utils" "context" "encoding/json" log "github.com/sirupsen/logrus" "math/rand" "time" ) // GQueryFunc 发送请求到客户端 func (c *Clients) GQueryFunc(funcName string, param string, resChan chan<- string) { if c.actionData[funcName] == nil { rwMu.Lock() c.actionData[funcName] = make(map[string]chan string) rwMu.Unlock() } var MessageId string for { MessageId = utils.GetUUID() if c.readFromMap(funcName, MessageId) == nil { rwMu.Lock() c.actionData[funcName][MessageId] = make(chan string, 1) rwMu.Unlock() break } utils.LogPrint("存在的消息id,跳过") } // 确保资源释放 defer func() { rwMu.Lock() delete(c.actionData[funcName], MessageId) rwMu.Unlock() close(resChan) }() // 构造消息并发送 WriteData := Message{Param: param, MessageId: MessageId, Action: funcName} data, err := json.Marshal(WriteData) if err != nil { log.Error(err, "JSON序列化失败") resChan <- "JSON序列化失败" return } rwMu.Lock() err = c.clientWs.WriteMessage(1, data) rwMu.Unlock() if err != nil { log.Error(err, "写入数据失败") resChan <- "rpc发送数据失败" return } // 使用 context 控制超时 ctx, cancel := context.WithTimeout(context.Background(), time.Duration(config.DefaultTimeout)*time.Second) defer cancel() resultChan := c.readFromMap(funcName, MessageId) if resultChan == nil { resChan <- "消息ID对应的管道不存在" return } select { case res := <-resultChan: resChan <- res case <-ctx.Done(): utils.LogPrint(MessageId + "超时了") resChan <- "获取结果超时 timeout" } } func getRandomClient(group string, clientId string) *Clients { var client *Clients // 不传递clientId时候,从group分组随便拿一个 if clientId != "" { clientName, ok := hlSyncMap.Load(group + "->" + clientId) if ok == false { return nil } client, _ = clientName.(*Clients) return client } groupClients := make([]*Clients, 0) //循环读取syncMap 获取group名字的 hlSyncMap.Range(func(_, value interface{}) bool { tmpClients, ok := value.(*Clients) if !ok { return true } if tmpClients.clientGroup == group { groupClients = append(groupClients, tmpClients) } return true }) if len(groupClients) == 0 { return nil } // 使用随机数发生器 r := rand.New(rand.NewSource(time.Now().UnixNano())) randomIndex := r.Intn(len(groupClients)) client = groupClients[randomIndex] return client }