feat: Optimizing RPC call (#2993)

* pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * rpc client * rpc client * rpc client * rpc client * rpc client * rpc client * rpc client * rpc client
2026-05-11 12:36:00 +08:00 · 2024-12-24 10:51:38 +08:00
parent e9895062cb
commit e3f609b3ae
67 changed files with 1287 additions and 1092 deletions
@@ -25,7 +25,6 @@ import (
 	"strconv"
 	"syscall"

-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/discovery/etcd"
 	"github.com/openimsdk/tools/utils/jsonutil"
@@ -93,10 +92,6 @@ func Start(ctx context.Context, index int, config *Config) error {
 	}
 	client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
-	if err = rpcclient.InitRpcCaller(client, config.Discovery.RpcService); err != nil {
-		return err
-	}
-
 	msgModel := redis.NewMsgCache(rdb)
 	msgDocModel, err := mgo.NewMsgMongo(mgocli.GetDB())
 	if err != nil {
@@ -116,7 +111,7 @@ func Start(ctx context.Context, index int, config *Config) error {
 	if err != nil {
 		return err
 	}
-	historyCH, err := NewOnlineHistoryRedisConsumerHandler(&config.KafkaConfig, msgTransferDatabase)
+	historyCH, err := NewOnlineHistoryRedisConsumerHandler(ctx, client, config, msgTransferDatabase)
 	if err != nil {
 		return err
 	}
@@ -18,6 +18,8 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcli"
+	"github.com/openimsdk/tools/discovery"
 	"strconv"
 	"strings"
 	"sync"
@@ -25,15 +27,12 @@ import (

 	"github.com/IBM/sarama"
 	"github.com/go-redis/redis"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/open-im-server/v3/pkg/tools/batcher"
 	"github.com/openimsdk/protocol/constant"
 	pbconv "github.com/openimsdk/protocol/conversation"
-	"github.com/openimsdk/protocol/group"
-	"github.com/openimsdk/protocol/rpccall"
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/errs"
 	"github.com/openimsdk/tools/log"
@@ -72,16 +71,30 @@ type OnlineHistoryRedisConsumerHandler struct {
 	msgTransferDatabase         controller.MsgTransferDatabase
 	conversationUserHasReadChan chan *userHasReadSeq
 	wg                          sync.WaitGroup
+
+	groupClient        *rpcli.GroupClient
+	conversationClient *rpcli.ConversationClient
 }

-func NewOnlineHistoryRedisConsumerHandler(kafkaConf *config.Kafka, database controller.MsgTransferDatabase) (*OnlineHistoryRedisConsumerHandler, error) {
+func NewOnlineHistoryRedisConsumerHandler(ctx context.Context, client discovery.SvcDiscoveryRegistry, config *Config, database controller.MsgTransferDatabase) (*OnlineHistoryRedisConsumerHandler, error) {
+	kafkaConf := config.KafkaConfig
 	historyConsumerGroup, err := kafka.NewMConsumerGroup(kafkaConf.Build(), kafkaConf.ToRedisGroupID, []string{kafkaConf.ToRedisTopic}, false)
 	if err != nil {
 		return nil, err
 	}
+	groupConn, err := client.GetConn(ctx, config.Discovery.RpcService.Group)
+	if err != nil {
+		return nil, err
+	}
+	conversationConn, err := client.GetConn(ctx, config.Discovery.RpcService.Conversation)
+	if err != nil {
+		return nil, err
+	}
 	var och OnlineHistoryRedisConsumerHandler
 	och.msgTransferDatabase = database
 	och.conversationUserHasReadChan = make(chan *userHasReadSeq, hasReadChanBuffer)
+	och.groupClient = rpcli.NewGroupClient(groupConn)
+	och.conversationClient = rpcli.NewConversationClient(conversationConn)
 	och.wg.Add(1)

 	b := batcher.New[sarama.ConsumerMessage](
@@ -109,15 +122,13 @@ func (och *OnlineHistoryRedisConsumerHandler) do(ctx context.Context, channelID
 	ctx = mcontext.WithTriggerIDContext(ctx, val.TriggerID())
 	ctxMessages := och.parseConsumerMessages(ctx, val.Val())
 	ctx = withAggregationCtx(ctx, ctxMessages)
-	log.ZInfo(ctx, "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMessages),
-		"key", val.Key())
+	log.ZInfo(ctx, "msg arrived channel", "channel id", channelID, "msgList length", len(ctxMessages), "key", val.Key())
 	och.doSetReadSeq(ctx, ctxMessages)

 	storageMsgList, notStorageMsgList, storageNotificationList, notStorageNotificationList :=
 		och.categorizeMessageLists(ctxMessages)
 	log.ZDebug(ctx, "number of categorized messages", "storageMsgList", len(storageMsgList), "notStorageMsgList",
-		len(notStorageMsgList), "storageNotificationList", len(storageNotificationList), "notStorageNotificationList",
-		len(notStorageNotificationList))
+		len(notStorageMsgList), "storageNotificationList", len(storageNotificationList), "notStorageNotificationList", len(notStorageNotificationList))

 	conversationIDMsg := msgprocessor.GetChatConversationIDByMsg(ctxMessages[0].message)
 	conversationIDNotification := msgprocessor.GetNotificationConversationIDByMsg(ctxMessages[0].message)
@@ -282,31 +293,26 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(ctx context.Context, key
 				log.ZDebug(ctx, "group chat first create conversation", "conversationID",
 					conversationID)

-				userIDs, err := rpccall.ExtractField(ctx, group.GetGroupMemberUserIDsCaller.Invoke,
-					&group.GetGroupMemberUserIDsReq{
-						GroupID: msg.GroupID,
-					}, (*group.GetGroupMemberUserIDsResp).GetUserIDs)
+				userIDs, err := och.groupClient.GetGroupMemberUserIDs(ctx, msg.GroupID)
 				if err != nil {
 					log.ZWarn(ctx, "get group member ids error", err, "conversationID",
 						conversationID)
 				} else {
 					log.ZInfo(ctx, "GetGroupMemberIDs end")

-					if err := pbconv.CreateGroupChatConversationsCaller.Execute(ctx, &pbconv.CreateGroupChatConversationsReq{
-						UserIDs: userIDs,
-						GroupID: msg.GroupID,
-					}); err != nil {
+					if err := och.conversationClient.CreateGroupChatConversations(ctx, msg.GroupID, userIDs); err != nil {
 						log.ZWarn(ctx, "single chat first create conversation error", err,
 							"conversationID", conversationID)
 					}
 				}
 			case constant.SingleChatType, constant.NotificationChatType:
-				if err := pbconv.CreateSingleChatConversationsCaller.Execute(ctx, &pbconv.CreateSingleChatConversationsReq{
+				req := &pbconv.CreateSingleChatConversationsReq{
 					RecvID:           msg.RecvID,
 					SendID:           msg.SendID,
 					ConversationID:   conversationID,
 					ConversationType: msg.SessionType,
-				}); err != nil {
+				}
+				if err := och.conversationClient.CreateSingleChatConversations(ctx, req); err != nil {
 					log.ZWarn(ctx, "single chat or notification first create conversation error", err,
 						"conversationID", conversationID, "sessionType", msg.SessionType)
 				}