feat 排队

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/coreos/go-semver v0.3.1 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.8 // indirect
@@ -49,6 +50,7 @@ require (
 	github.com/natefinch/lumberjack v2.0.0+incompatible // indirect
 	github.com/panjf2000/gnet/v2 v2.9.7 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/quic-go/qpack v0.5.1 // indirect
 	github.com/quic-go/quic-go v0.54.0 // indirect
 	github.com/redis/go-redis/v9 v9.10.0 // indirect
@@ -58,6 +60,7 @@ require (
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
 	github.com/spf13/viper v1.21.0 // indirect
+	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.3.0 // indirect
@@ -79,6 +82,7 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gorm.io/datatypes v1.2.4 // indirect
 	gorm.io/driver/mysql v1.6.0 // indirect
 	gorm.io/hints v1.1.0 // indirect

go.sum

@@ -39,6 +39,8 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
+github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
 github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
 github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
@@ -111,8 +113,7 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/natefinch/lumberjack v2.0.0+incompatible h1:4QJd3OLAMgj7ph+yZTuX13Ld4UpgHp07nNdFX7mqFfM=
 github.com/natefinch/lumberjack v2.0.0+incompatible/go.mod h1:Wi9p2TTF5DG5oU+6YfsmYQpsTIOm0B1VNzQg9Mw6nPk=
-github.com/panjf2000/gnet/v2 v2.9.1 h1:bKewICy/0xnQ9PMzNaswpe/Ah14w1TrRk91LHTcbIlA=
+github.com/panjf2000/gnet/v2 v2.9.7 h1:6zW7Jl3oAfXwSuh1PxHLndoL2MQRWx0AJR6aaQjxUgA=
-github.com/panjf2000/gnet/v2 v2.9.1/go.mod h1:WQTxDWYuQ/hz3eccH0FN32IVuvZ19HewEWx0l62fx7E=
 github.com/panjf2000/gnet/v2 v2.9.7/go.mod h1:WQTxDWYuQ/hz3eccH0FN32IVuvZ19HewEWx0l62fx7E=
 github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=

internal/grpc_server/stream_client/gateway.go

@@ -6,60 +6,79 @@ import (
 	"context"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
+	"strconv"
+	"sync"
 )
 
-var gatewayServerM map[int64]map[GatewayFun]grpc.ClientStream // map[sid]map[方法名]流连接
-
 type GatewayFun int
 
 const (
 	FunToClient GatewayFun = iota
 )
 
-func init() {
+var gatewayServer sync.Map // map[string]*gatewayStream
-	gatewayServerM = make(map[int64]map[GatewayFun]grpc.ClientStream)
+
+type gatewayStream struct {
+	mu     sync.Mutex
+	stream grpc.ClientStream
 }
 
-func FindGatewayBySID(sid int64, fun GatewayFun) (grpc.ClientStream, error) {
+func findGatewayBySID(sid int64, fun GatewayFun) (*gatewayStream, error) {
-	g := gatewayServerM[sid]
+	key := gatewayKey(sid, fun)
-	if g == nil {
+
-		g = make(map[GatewayFun]grpc.ClientStream)
+	if v, ok := gatewayServer.Load(key); ok {
-		gatewayServerM[sid] = g
+		return v.(*gatewayStream), nil
 	}
-	gatewayLink := g[fun]
+
-	if gatewayLink == nil {
+	client, err := service.GatewayNewClient(sid)
-		gatewayClient, err := service.GatewayNewClient(sid)
+	if err != nil {
-		if err != nil {
+		log.Errorf("findGatewayBySID cannot find client: %v", err)
-			log.Errorf("cannot find gatewayClient: %v", err)
+		return nil, err
-			return nil, err
-		}
-		var link grpc.ClientStream
-		switch fun {
-		case FunToClient:
-			link, err = gatewayClient.ToClient(context.Background())
-		}
-		if err != nil {
-			log.Errorf("FindGatewayBySID %v err: %v, sid: %v", fun, err, sid)
-			return nil, err
-		}
-		g[fun] = link
-		gatewayLink = link
 	}
-	return gatewayLink, nil
+	var stream grpc.ClientStream
+	switch fun {
+	case FunToClient:
+		stream, err = client.ToClient(context.Background())
+	}
+	if err != nil {
+		log.Errorf("findGatewayBySID %v err: %v, sid: %v", fun, err, sid)
+		return nil, err
+	}
+
+	ss := &gatewayStream{stream: stream}
+	if actual, loaded := gatewayServer.LoadOrStore(key, ss); loaded {
+		go func() { _ = stream.CloseSend() }()
+		return actual.(*gatewayStream), nil
+	}
+
+	return ss, nil
 }
 
 func SendMessageToGateway(sid int64, fun GatewayFun, msg proto.Message, re ...bool) error {
-	stream, err := FindGatewayBySID(sid, fun)
+	ss, err := findGatewayBySID(sid, fun)
 	if err != nil {
 		return err
 	}
-	if err = stream.SendMsg(msg); err != nil {
+
+	ss.mu.Lock()
+	err = ss.stream.SendMsg(msg)
+	ss.mu.Unlock()
+
+	if err != nil {
+		key := gatewayKey(sid, fun)
+		if v, ok := gatewayServer.Load(key); ok && v == ss {
+			gatewayServer.Delete(key)
+			_ = ss.stream.CloseSend()
+		}
+		// 如果没有标识本次是重试的，就重试一次（默认重试）
 		if re == nil || !re[0] {
-			_ = stream.CloseSend()
-			delete(gatewayServerM[sid], fun)
 			return SendMessageToGateway(sid, fun, msg, true)
 		}
 		return err
 	}
 	return nil
 }
+
+func gatewayKey(sid int64, fun GatewayFun) string {
+	return strconv.FormatInt(sid, 10) + "-" + strconv.Itoa(int(fun))
+}