perf: broadcast msg to all gateway with concurrency (#1411)

Signed-off-by: rfyiamcool <rfyiamcool@163.com>

internal/push/push_to_client.go

@@ -18,8 +18,9 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"sync"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/conversation"
@@ -40,6 +41,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
@@ -285,18 +287,44 @@ func (p *Pusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
 	if err != nil {
 		return nil, err
 	}
+
+	var (
+		mu         sync.Mutex
+		wg         = errgroup.Group{}
+		input      = &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs}
+		maxWorkers = config.Config.Push.MaxConcurrentWorkers
+	)
+
+	if maxWorkers < 3 {
+		maxWorkers = 3
+	}
+
+	wg.SetLimit(maxWorkers)
+
 	// Online push message
-	for _, v := range conns {
+	for _, conn := range conns {
-		msgClient := msggateway.NewMsgGatewayClient(v)
+		conn := conn // loop var safe
-		reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs})
+		wg.Go(func() error {
-		if err != nil {
+			msgClient := msggateway.NewMsgGatewayClient(conn)
-			continue
+			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
-		}
+			if err != nil {
-		log.ZDebug(ctx, "push result", "reply", reply)
+				return nil
-		if reply != nil && reply.SinglePushResult != nil {
+			}
-			wsResults = append(wsResults, reply.SinglePushResult...)
+
-		}
+			log.ZDebug(ctx, "push result", "reply", reply)
+			if reply != nil && reply.SinglePushResult != nil {
+				mu.Lock()
+				wsResults = append(wsResults, reply.SinglePushResult...)
+				mu.Unlock()
+			}
+
+			return nil
+		})
 	}
+
+	_ = wg.Wait()
+
+	// always return nil
 	return wsResults, nil
 }
 

pkg/common/config/config.go

@@ -199,8 +199,9 @@ type configStruct struct {
 	} `yaml:"longConnSvr"`
 
 	Push struct {
-		Enable string `yaml:"enable"`
+		MaxConcurrentWorkers int    `yaml:"maxConcurrentWorkers"`
-		GeTui  struct {
+		Enable               string `yaml:"enable"`
+		GeTui                struct {
 			PushUrl      string `yaml:"pushUrl"`
 			AppKey       string `yaml:"appKey"`
 			Intent       string `yaml:"intent"`