fix: memory queue optimization (#2568)

* fix: GroupApplicationAcceptedNotification * fix: GroupApplicationAcceptedNotification * fix: NotificationUserInfoUpdate * cicd: robot automated Change * fix: component * fix: getConversationInfo * feat: cron task * feat: cron task * feat: cron task * feat: cron task * feat: cron task * fix: minio config url recognition error * update gomake version * update gomake version * fix: seq conversion bug * fix: redis pipe exec * fix: ImportFriends * fix: A large number of logs keysAndValues length is not even * feat: mark read aggregate write * feat: online status supports redis cluster * feat: online status supports redis cluster * feat: online status supports redis cluster * merge * merge * read seq is written to mongo * read seq is written to mongo * fix: invitation to join group notification * fix: friend op_user_id * feat: optimizing asynchronous context * feat: optimizing memamq size --------- Co-authored-by: withchao <withchao@users.noreply.github.com>
openimsdk · Aug 29, 2024 · 275491a · 275491a
1 parent c37e022
commit 275491a
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 6 deletions.
diff --git a/internal/rpc/relation/friend.go b/internal/rpc/relation/friend.go
@@ -121,7 +121,7 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
 		conversationRpcClient: rpcclient.NewConversationRpcClient(client, config.Share.RpcRegisterName.Conversation),
 		config:                config,
 		webhookClient:         webhook.NewWebhookClient(config.WebhooksConfig.URL),
-		queue:                 memamq.NewMemoryQueue(128, 1024*8),
+		queue:                 memamq.NewMemoryQueue(16, 1024*1024),
 	})
 	return nil
 }

diff --git a/pkg/common/storage/controller/friend.go b/pkg/common/storage/controller/friend.go
@@ -160,7 +160,7 @@ func (f *friendDatabase) BecomeFriends(ctx context.Context, ownerUserID string,
 		if err != nil {
 			return err
 		}
-		opUserID := mcontext.GetOperationID(ctx)
+		opUserID := mcontext.GetOpUserID(ctx)
 		friends := make([]*model.Friend, 0, len(friendUserIDs)*2)
 		myFriendsSet := datautil.SliceSetAny(myFriends, func(friend *model.Friend) string {
 			return friend.FriendUserID

diff --git a/pkg/rpcclient/msg.go b/pkg/rpcclient/msg.go
@@ -23,7 +23,6 @@ import (
 	"github.com/openimsdk/protocol/sdkws"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/log"
-	"github.com/openimsdk/tools/mcontext"
 	"github.com/openimsdk/tools/mq/memamq"
 	"github.com/openimsdk/tools/system/program"
 	"github.com/openimsdk/tools/utils/idutil"
@@ -270,8 +269,8 @@ func WithUserRpcClient(userRpcClient *UserRpcClient) NotificationSenderOptions {
 }
 
 const (
-	notificationWorkerCount = 2
-	notificationBufferSize  = 200
+	notificationWorkerCount = 16
+	notificationBufferSize  = 1024 * 1024 * 2
 )
 
 func NewNotificationSender(conf *config.Notification, opts ...NotificationSenderOptions) *NotificationSender {
@@ -298,7 +297,8 @@ func WithRpcGetUserName() NotificationOptions {
 }
 
 func (s *NotificationSender) send(ctx context.Context, sendID, recvID string, contentType, sessionType int32, m proto.Message, opts ...NotificationOptions) {
-	ctx = mcontext.WithMustInfoCtx([]string{mcontext.GetOperationID(ctx), mcontext.GetOpUserID(ctx), mcontext.GetOpUserPlatform(ctx), mcontext.GetConnID(ctx)})
+	//ctx = mcontext.WithMustInfoCtx([]string{mcontext.GetOperationID(ctx), mcontext.GetOpUserID(ctx), mcontext.GetOpUserPlatform(ctx), mcontext.GetConnID(ctx)})
+	ctx = context.WithoutCancel(ctx)
 	ctx, cancel := context.WithTimeout(ctx, time.Second*time.Duration(5))
 	defer cancel()
 	n := sdkws.NotificationElem{Detail: jsonutil.StructToJsonString(m)}