send message parallel

internal/rpc/msg/send_msg.go

@@ -18,6 +18,7 @@ import (
 	"math/rand"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -255,7 +256,78 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 			}
 		case constant.MemberQuitNotification:
 			addUidList = append(addUidList, pb.MsgData.SendID)
-		case constant.AtText:
+
+		default:
+		}
+		groupID := pb.MsgData.GroupID
+		//split  parallel send
+		var wg sync.WaitGroup
+		var sendTag bool
+		var split = 50
+		remain := len(memberUserIDList) % split
+		for i := 0; i < len(memberUserIDList)/split; i++ {
+			wg.Add(1)
+			go func(list []string) {
+				for _, v := range list {
+					pb.MsgData.RecvID = v
+					isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
+					if isSend {
+						msgToMQ.MsgData = pb.MsgData
+						err := rpc.sendMsgToKafka(&msgToMQ, v)
+						if err != nil {
+							log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
+						} else {
+							sendTag = true
+						}
+					}
+				}
+				wg.Done()
+			}(memberUserIDList[i*split : (i+1)*split])
+		}
+		if remain > 0 {
+			wg.Add(1)
+			go func(list []string) {
+				for _, v := range list {
+					pb.MsgData.RecvID = v
+					isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
+					if isSend {
+						msgToMQ.MsgData = pb.MsgData
+						err := rpc.sendMsgToKafka(&msgToMQ, v)
+						if err != nil {
+							log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
+						} else {
+							sendTag = true
+						}
+					}
+				}
+				wg.Done()
+			}(memberUserIDList[split*(len(memberUserIDList)/split):])
+		}
+		wg.Wait()
+
+		log.Info(msgToMQ.OperationID, "addUidList", addUidList)
+		for _, v := range addUidList {
+			pb.MsgData.RecvID = v
+			isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
+			log.Info(msgToMQ.OperationID, "isSend", isSend)
+			if isSend {
+				msgToMQ.MsgData = pb.MsgData
+				err := rpc.sendMsgToKafka(&msgToMQ, v)
+				if err != nil {
+					log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
+				} else {
+					sendTag = true
+				}
+			}
+		}
+		// callback
+		if err := callbackAfterSendGroupMsg(pb); err != nil {
+			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendGroupMsg failed", err.Error())
+		}
+		if !sendTag {
+			return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
+		} else {
+			if pb.MsgData.ContentType == constant.AtText {
 				go func() {
 					var conversationReq pbConversation.ModifyConversationFieldReq
 					var tag bool
@@ -305,41 +377,10 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 						}
 					}
 				}()
-		default:
-		}
-		groupID := pb.MsgData.GroupID
-		for _, v := range memberUserIDList {
-			pb.MsgData.RecvID = v
-			isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
-			if isSend {
-				msgToMQ.MsgData = pb.MsgData
-				err := rpc.sendMsgToKafka(&msgToMQ, v)
-				if err != nil {
-					log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
-					return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
-				}
 			}
+			return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime)
 
 		}
-		log.Info(msgToMQ.OperationID, "addUidList", addUidList)
-		for _, v := range addUidList {
-			pb.MsgData.RecvID = v
-			isSend := modifyMessageByUserMessageReceiveOpt(v, groupID, constant.GroupChatType, pb)
-			log.Info(msgToMQ.OperationID, "isSend", isSend)
-			if isSend {
-				msgToMQ.MsgData = pb.MsgData
-				err := rpc.sendMsgToKafka(&msgToMQ, v)
-				if err != nil {
-					log.NewError(msgToMQ.OperationID, "kafka send msg err:UserId", v, msgToMQ.String())
-					return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
-				}
-			}
-		}
-		// callback
-		if err := callbackAfterSendGroupMsg(pb); err != nil {
-			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendGroupMsg failed", err.Error())
-		}
-		return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime)
 	case constant.NotificationChatType:
 		msgToMQ.MsgData = pb.MsgData
 		log.NewInfo(msgToMQ.OperationID, msgToMQ)