负载同步

game/config/data/ServerConfig.json

@@ -0,0 +1,3 @@
+[
+		{"ID":1,"Name":"加权随机规则","Enable":1,"Max":20000,"Expansion":15000,"Weight":{"Min":1,"Middle":3,"Max":10},"Range":100}
+]

game/game_cluster/internal/data/ServerConfig.go

@@ -0,0 +1,65 @@
+// this file is auto create by program, don't edit manually
+
+package data
+
+import (
+	mhayaError "github.com/mhaya/error"
+	mhayaLogger "github.com/mhaya/logger"
+)
+
+type serverConfig struct {
+	maps map[int]*ServerConfigRow
+}
+
+type ServerConfigRow struct {
+	ID        int         // 配置ID
+	Name      string      // 配置描述信息
+	Enable    int         // 是否启用0否1是
+	Max       int         // 服务器最大负载
+	Expansion int         // 扩容指标（到达扩容指标权重降低到1）
+	Weight    WeightInt32 // 权重规则
+	Range     int         // 规则标准(均值每100为标准)
+}
+
+func (p *serverConfig) Name() string {
+	return "ServerConfig"
+}
+
+func (p *serverConfig) Init() {
+	p.maps = make(map[int]*ServerConfigRow)
+}
+
+func (p *serverConfig) OnLoad(maps interface{}, _ bool) (int, error) {
+	list, ok := maps.([]interface{})
+	if !ok {
+		return 0, mhayaError.Error("maps convert to []interface{} error.")
+	}
+
+	loadMaps := make(map[int]*ServerConfigRow)
+	for index, data := range list {
+		loadConfig := &ServerConfigRow{}
+		err := DecodeData(data, loadConfig)
+		if err != nil {
+			mhayaLogger.Warnf("decode error. [row = %d, %v], err = %s", index+1, loadConfig, err)
+			continue
+		}
+
+		loadMaps[loadConfig.ID] = loadConfig
+	}
+
+	p.maps = loadMaps
+
+	return len(list), nil
+}
+
+func (p *serverConfig) OnAfterLoad(_ bool) {
+}
+
+func (p *serverConfig) Get() (*ServerConfigRow, bool) {
+	for _, row := range p.maps {
+		if row.Enable == 1 {
+			return row, true
+		}
+	}
+	return nil, false
+}

game/game_cluster/internal/data/TypeDefine.go

@@ -147,3 +147,9 @@ type Vector2 struct {
 	X int // 存储一个二维空间的坐标
 	Y int //
 }
+
+type WeightInt32 struct {
+	Min    int // 描述一个值的范围
+	Middle int //
+	Max    int //
+}

game/game_cluster/internal/data/component.go

@@ -28,6 +28,7 @@ var (
 	ContryConfig       = &contryConfig{}
 	PlayerConfig       = &playerConfig{}
 	KolConfig          = &kolConfig{}
+	ServerConfig       = &serverConfig{}
 )
 
 func New() *mhayaDataConfig.Component {
@@ -52,6 +53,7 @@ func New() *mhayaDataConfig.Component {
 		ContryConfig,
 		PlayerConfig,
 		KolConfig,
+		ServerConfig,
 	)
 	return dataConfig
 }

game/game_cluster/internal/mdb/component.go

@@ -1,7 +1,6 @@
 package mdb
 
 import (
-	"crypto/tls"
 	"github.com/go-redis/redis/v8"
 	mhayaMongo "github.com/mhaya/components/mongo"
 	clog "github.com/mhaya/logger"
@@ -37,15 +36,15 @@ func (p *ActorDB) OnInit() {
 		clog.Panic("game_db_id not found")
 	}
 	redisConfig := cprofile.GetConfig("redis")
-	tlsConfig := &tls.Config{
+	/*	tlsConfig := &tls.Config{
 		MinVersion:               tls.VersionTLS12,
 		PreferServerCipherSuites: true,
-	}
+	}*/
 	RDB = redis.NewUniversalClient(&redis.UniversalOptions{
-		Addrs:     []string{redisConfig.GetString("address")},
-		Password:  redisConfig.GetString("password"),
-		DB:        redisConfig.GetInt("db"),
-		TLSConfig: tlsConfig,
+		Addrs:    []string{redisConfig.GetString("address")},
+		Password: redisConfig.GetString("password"),
+		DB:       redisConfig.GetInt("db"),
+		//TLSConfig: tlsConfig,
 	})
 
 	if p.App().NodeId() == "m-center" {

game/game_cluster/nodes/center/module/account/actor_account.go

@@ -34,10 +34,10 @@ func (p *ActorAccount) AliasID() string {
 // OnInit center为后端节点，不直接与客户端通信，所以注册了一些remote函数，供RPC调用
 func (p *ActorAccount) OnInit() {
 	p.Remote().Register("registerAccount", p.registerOrLoinAccount)
-	p.Timer().Add(5*time.Second, p.load)
+	p.Timer().Add(30*time.Second, p.load)
 	p.Timer().Add(time.Minute, p.Stat)
 	p.Timer().Add(1*time.Minute, p.PlayerIpStat)
-	p.Timer().Add(time.Minute, p.playerLevelStat)
+	//	p.Timer().Add(time.Minute, p.playerLevelStat)
 	p.Timer().AddFixedHour(23, 59, 59, p.Stat)
 	//p.Timer().AddOnce(30*time.Second, RunToMongoDB)
 	// p.Timer().AddOnce(30*time.Second, RunToMongoDB)
@@ -52,7 +52,7 @@ func (p *ActorAccount) Stat() {
 	channelConfig := data.ChannelConfig.GatMap()
 	for _, v := range platformConfig {
 		for _, v2 := range channelConfig {
-			p.dailyStat(v.Name, v2.Name, daily)
+			go p.dailyStat(v.Name, v2.Name, daily)
 		}
 	}
 }
@@ -62,7 +62,7 @@ func (p *ActorAccount) playerLevelStat() {
 	channelConfig := data.ChannelConfig.GatMap()
 	for _, v := range platformConfig {
 		for _, v2 := range channelConfig {
-			p.levelStat(v.Name, v2.Name)
+			go p.levelStat(v.Name, v2.Name)
 		}
 	}
 }
@@ -73,7 +73,7 @@ func (p *ActorAccount) PlayerIpStat() {
 	channelConfig := data.ChannelConfig.GatMap()
 	for _, v := range platformConfig {
 		for _, v2 := range channelConfig {
-			p.ipStat(v.Name, v2.Name, daily)
+			go p.ipStat(v.Name, v2.Name, daily)
 		}
 	}
 }

game/game_cluster/nodes/web/controller/controller.go

@@ -128,7 +128,34 @@ func (p *Controller) Auth(c *mhayaGin.Context) (*token.Token, int32) {
 	if oldToken != tokenString {
 		return nil, code.AccountTokenValidateFail
 	}
-	return userToken, code.OK
+
+	sdkInvoke, err := sdk.GetInvoke(userToken.PID)
+	if err != nil {
+		mhayaLogger.Warnf("[Auth] [pid = %d] get invoke error. params=%v", userToken.PID, userToken)
+		return nil, code.AccountTokenValidateFail
+	}
+
+	s, newToken := sdkInvoke.Reconnect(userToken)
+	if code.IsFail(s) {
+		mhayaLogger.Warnf("[Auth]  Reconnect err , errCode = %v", s)
+		return nil, code.AccountTokenValidateFail
+	}
+	//如果重新刷新路由
+	if newToken.TargetPath != userToken.TargetPath {
+		config := data.SdkConfig.Get(newToken.PID)
+		if config == nil {
+			mhayaLogger.Warnf("[Auth] if config err.  params=%v", newToken)
+			return nil, code.AccountTokenValidateFail
+		}
+		base64Token := token.New(userToken.PID, userToken.OpenID, userToken.PlayerID, newToken.TargetPath, "", config.Salt).ToBase64()
+		err = mdb.RDB.Set(context.Background(), fmt.Sprintf("%v:%v", constant.Token, userToken.OpenID), base64Token, mhayaTime.SecondsPerWeek*time.Second).Err()
+		if err != nil {
+			mhayaLogger.Warnf("[Auth] result set token err. result = %s", newToken)
+			return nil, code.AccountTokenValidateFail
+		}
+	}
+
+	return newToken, code.OK
 }
 
 // login 根据pid获取sdkConfig，与第三方进行帐号登陆效验

game/game_cluster/nodes/web/sdk/quick_sdk.go

@@ -10,6 +10,7 @@ import (
 	"github.com/mhaya/game/game_cluster/internal/mdb/models"
 	"github.com/mhaya/game/game_cluster/internal/param"
 	sessionKey "github.com/mhaya/game/game_cluster/internal/session_key"
+	"github.com/mhaya/game/game_cluster/internal/token"
 	clog "github.com/mhaya/logger"
 )
 
@@ -81,7 +82,33 @@ func (p quickSdk) Login(config *data.SdkConfigRow, params Params, callback Callb
 	})
 }
 
-func (s quickSdk) PayCallback(config *data.SdkConfigRow, c *mhayaGin.Context) {
+func (p quickSdk) Reconnect(token *token.Token) (int32, *token.Token) {
+	list := p.app.Discovery().ListByType("game")
+	targetPath, err := cfacade.ToActorPath(token.TargetPath)
+	if err != nil {
+		clog.Warnf("[Reconnect] Target path error. targetPath = %v, error = %v", token.TargetPath, err)
+		return code.AccountAuthFail, nil
+	}
+	isNew := false
+	for _, v := range list {
+		if v.GetNodeId() == targetPath.NodeID {
+			isNew = true
+			break
+		}
+	}
+	if !isNew {
+		nodeId, ok := SetNode(p.app)
+		if code.IsFail(ok) {
+			clog.Warnf("[RegisterAccount] openID = %s,nodeID=%v, errCode = %v", token.OpenID, nodeId, err)
+			return code.AccountAuthFail, nil
+		}
+		path := cfacade.NewChildPath(nodeId, "player", token.PlayerID)
+		token.TargetPath = path
+	}
+	return code.OK, token
+}
+
+func (p quickSdk) PayCallback(config *data.SdkConfigRow, c *mhayaGin.Context) {
 	// TODO 这里实现quick sdk 支付回调的逻辑
 	c.RenderHTML("FAIL")
 }

game/game_cluster/nodes/web/sdk/sdk.go

@@ -10,7 +10,8 @@ import (
 	"github.com/mhaya/game/game_cluster/internal/constant"
 	"github.com/mhaya/game/game_cluster/internal/data"
 	"github.com/mhaya/game/game_cluster/internal/mdb"
-	"math"
+	"github.com/mhaya/game/game_cluster/internal/token"
+	mhayaLogger "github.com/mhaya/logger"
 	"math/rand"
 )
 
@@ -27,6 +28,7 @@ var (
 type (
 	Invoke interface {
 		SdkId() int                                                        // sdk id
+		Reconnect(token *token.Token) (int32, *token.Token)                //重连
 		Login(config *data.SdkConfigRow, params Params, callback Callback) // Login 登录验证接口
 		PayCallback(config *data.SdkConfigRow, c *mhayaGin.Context)        // PayCallback 支付回调接口
 	}
@@ -101,19 +103,25 @@ func SetNode(app cfacade.IApplication) (string, int32) {
 		if len(m) == 0 {
 			nodeId = list[key].GetNodeId()
 		} else {
-			num := math.MaxInt
+			var maps = make(map[string]int, 10)
 			for _, v := range list {
-				id := v.GetNodeId()
-				total, ok := mhayaString.ToInt(m[id])
-				if ok {
-					if total < num {
-						num = total
-						nodeId = v.GetNodeId()
+				if d, ok := m[v.GetNodeId()]; ok {
+					p, ok := mhayaString.ToInt(d)
+					if !ok {
+						nodeId = list[key].GetNodeId()
 					}
+					maps[v.GetNodeId()] = p
 				} else {
-					mdb.RDB.HSet(context.Background(), constant.ServerLoadHKey, id, 0)
+					maps[v.GetNodeId()] = 0
 				}
 			}
+			wwr := NewWeightedRoundRobin(maps)
+			info := wwr.Pick()
+			if info == nil {
+				mhayaLogger.Warnf("[SetNode] Load too high. data=%v", maps)
+				return "", code.Error
+			}
+			nodeId = info.NodeId
 		}
 	}
 	return nodeId, code.OK

game/game_cluster/nodes/web/sdk/sdk_pg.go

@@ -1 +0,0 @@
-package sdk

game/game_cluster/nodes/web/sdk/serverload.go

@@ -0,0 +1,86 @@
+package sdk
+
+import (
+	"github.com/mhaya/game/game_cluster/internal/data"
+	mhayaLogger "github.com/mhaya/logger"
+	"math/rand"
+)
+
+// Service 结构体表示单个服务的信息。
+type Service struct {
+	NodeId string // 服务的节点地址
+	Player int
+	Weight float64 // 权重
+}
+
+// WeightedRoundRobin 是一个结构体，用于管理一组带权重的服务，并支持基于权重的选择逻辑。
+type WeightedRoundRobin struct {
+	services []*Service
+	all      int
+	total    float64 // 所有服务的总权重
+}
+
+// NewWeightedRoundRobin 创建一个新的 WeightedRoundRobin 实例。
+func NewWeightedRoundRobin(maps map[string]int) *WeightedRoundRobin {
+	serviceConfig, ok := data.ServerConfig.Get()
+	if !ok {
+		mhayaLogger.Warnf("[NewWeightedRoundRobin] weighted round robin service config is nil")
+		return nil
+	}
+	wrr := &WeightedRoundRobin{
+		services: nil,
+	}
+	var services []*Service
+	for k, v := range maps {
+		wrr.all += v
+		services = append(services, &Service{
+			NodeId: k,
+			Player: v,
+		})
+	}
+	wrr.services = services
+	avg := wrr.all / len(wrr.services)
+
+	for _, service := range services {
+		if service.Player-avg <= 3*serviceConfig.Range {
+			service.Weight = float64(serviceConfig.Weight.Max)
+		} else if service.Player-avg < 2*serviceConfig.Range {
+			service.Weight = float64(serviceConfig.Weight.Middle)
+		} else {
+			service.Weight = float64(serviceConfig.Weight.Min)
+		}
+		if service.Player >= serviceConfig.Expansion {
+			service.Weight = float64(serviceConfig.Weight.Min)
+		}
+		if service.Player >= serviceConfig.Max {
+			service.Weight = 0
+		}
+	}
+
+	for _, s := range services {
+		wrr.total += s.Weight
+	}
+	return wrr
+}
+
+// Pick 使用加权轮询的方式选取一个服务。
+func (w *WeightedRoundRobin) Pick() *Service {
+	if len(w.services) == 0 {
+		return nil
+	}
+
+	// 随机选择一个介于[0, total)之间的数字
+	target := rand.Float64() * w.total
+	var current float64 = 0.0
+
+	// 循环直到找到正确的服务
+	for _, service := range w.services {
+		current += service.Weight
+		if current >= target {
+			return service
+		}
+	}
+
+	// 如果因浮点数精度问题未能命中，则返回最后一个服务
+	return w.services[len(w.services)-1]
+}

game/game_cluster/robot_client/main.go

@@ -18,8 +18,8 @@ func main() {
 
 	// 定义命令行标志
 	userOffset := flag.Int("userOffset", 0, "用户偏移量")
-	count := flag.Int("count", 20000, "请求总数")
-	qps := flag.Int("qps", 50, "最大并发数")
+	count := flag.Int("count", 80000, "请求总数")
+	qps := flag.Int("qps", 500, "最大并发数")
 	url := flag.String("url", "http://127.0.0.1:20000", "请求url")
 	url1 := flag.String("url1", "http://127.0.0.1:20001", "请求url")
 	url2 := flag.String("url2", "http://127.0.0.1:20002", "请求url")