负载同步

components/mongo/component.go

@@ -88,7 +88,7 @@ func (s *Component) Init() {
 }
 
 func CreateDatabase(uri, dbName string, timeout ...time.Duration) (*mongo.Database, error) {
-	tt := 3 * time.Second
+	tt := 5 * time.Second
 
 	if len(timeout) > 0 && timeout[0].Seconds() > 3 {
 		tt = timeout[0]
@@ -99,7 +99,7 @@ func CreateDatabase(uri, dbName string, timeout ...time.Duration) (*mongo.Databa
 	//}
 	//o := options.Client().ApplyURI(uri).SetTLSConfig(tlsConfig)
 
-	o := options.Client().ApplyURI(uri)
+	o := options.Client().ApplyURI(uri).SetMaxPoolSize(500)
 	if err := o.Validate(); err != nil {
 		return nil, err
 	}

game/game_cluster/internal/constant/constant.go

@@ -28,6 +28,7 @@ const (
 	MoneyRatio          = 10000000
 
 	AdminAccess = "admin"
+	NewPlayers  = "newPlayers"
 
 	TotalIncomeKey = "TotalIncome"
 )

game/game_cluster/internal/mdb/models/account.go

@@ -2,6 +2,8 @@ package models
 
 import (
 	"context"
+	"fmt"
+	"github.com/go-redis/redis/v8"
 	jsoniter "github.com/json-iterator/go"
 	mhayaTime "github.com/mhaya/extend/time"
 	"github.com/mhaya/game/game_cluster/internal/code"
@@ -15,6 +17,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 	"sort"
+	"time"
 )
 
 type Account struct {
@@ -49,11 +52,24 @@ type RocketLvProgress struct {
 
 func (ac *Account) AccountRegisterOrLogin(req *param.LoginReq) (*Account, int32) {
 	var account Account
+	ctx := context.Background()
+	acc, err := mdb.RDB.Get(ctx, fmt.Sprintf("%v:%v", constant.CNameAccount, req.OpenID)).Bytes()
+	if err != nil && err != redis.Nil {
+		clog.Debugf("[Account] AccountRegisterOrLogin error: %v", err)
+	}
 	findFilter := bson.M{"openId": req.OpenID}
-	err := mdb.MDB.Collection(constant.CNameAccount).FindOne(context.Background(), findFilter).Decode(&account)
-	if err != nil && err != mongo.ErrNoDocuments {
-		clog.Errorf("Failed to AccountRegisterOrLogin select err request: %v  err = %v", req, err.Error())
-		return nil, code.LoginError
+	if len(acc) == 0 {
+		err = mdb.MDB.Collection(constant.CNameAccount).FindOne(context.Background(), findFilter).Decode(&account)
+		if err != nil && err != mongo.ErrNoDocuments {
+			clog.Errorf("Failed to AccountRegisterOrLogin select err request: %v  err = %v", req, err.Error())
+			return nil, code.LoginError
+		}
+	} else {
+		err = jsoniter.Unmarshal(acc, &account)
+		if err != nil {
+			clog.Debugf("[Account] unmarshal AccountRegisterOrLogin data error: %v", err)
+			return nil, code.LoginError
+		}
 	}
 
 	// todo 测试代码
@@ -83,6 +99,9 @@ func (ac *Account) AccountRegisterOrLogin(req *param.LoginReq) (*Account, int32)
 			account.LoginIp = req.IP
 			account.Platform = req.Platform
 			mdb.MDB.Collection(constant.CNameAccount).UpdateOne(context.Background(), findFilter, bson.M{"$set": bson.M{"platform": req.Platform, "account": user.Username, "isPremium": user.IsPremium, "loginIp": req.IP}})
+
+			jsonData, _ := jsoniter.Marshal(&account)
+			err = mdb.RDB.Set(context.Background(), fmt.Sprintf("%v:%v", constant.CNameAccount, account.OpenId), jsonData, time.Hour*3*24).Err()
 		}
 
 		return &account, code.OK
@@ -100,15 +119,15 @@ func (ac *Account) AccountRegisterOrLogin(req *param.LoginReq) (*Account, int32)
 		JoinTime:  mhayaTime.Now().Unix(),
 	}
 
-	accountStr, _ := jsoniter.MarshalToString(devAccountTable)
-	err = mdb.RDB.Set(context.Background(), constant.CNameAccount, accountStr, 0).Err()
-
 	_, err = mdb.MDB.Collection(constant.CNameAccount).InsertOne(context.Background(), devAccountTable)
 	if err != nil {
 		clog.Errorf("Failed to AccountRegisterOrLogin request: %v  err = %v", devAccountTable, err.Error())
 		return nil, code.LoginError
 	}
 
+	jsonData, _ := jsoniter.Marshal(&devAccountTable)
+	err = mdb.RDB.Set(context.Background(), fmt.Sprintf("%v:%v", constant.CNameAccount, devAccountTable.OpenId), jsonData, time.Hour*3*24).Err()
+
 	//统计新注册
 	SetDailyRecordNewUserRegisterHash(req.Platform, req.Channel, devAccountTable.UserName, req.IP, DailyRecordNewRegistered)
 

game/game_cluster/internal/mdb/mongo_index.go

@@ -66,36 +66,13 @@ func SetCashOutRecordIndex() error {
 
 func SetPlayerIndex() error {
 
-	keys := bson.D{
-		{"nickName", 1},
-	}
-	indexModel := mongo.IndexModel{
-		Keys: keys,
-	}
-	err := Index(indexModel, constant.CNamePlayer)
-	if err != nil {
-		return err
-	}
-
-	keysUnique := bson.D{
-		{"userName", 1},
-	}
-	indexModelUnique1 := mongo.IndexModel{
-		Keys:    keysUnique,
-		Options: options.Index().SetUnique(true), // 设置索引选项，例如唯一索引
-	}
-	err = Unique(indexModelUnique1, constant.CNamePlayer)
-	if err != nil {
-		return err
-	}
 	keysUnique1 := bson.D{
 		{"openId", 1},
 	}
 	indexModelUnique := mongo.IndexModel{
-		Keys:    keysUnique1,                     // 1 表示升序，-1 表示降序
-		Options: options.Index().SetUnique(true), // 设置索引选项，例如唯一索引
+		Keys: keysUnique1, // 1 表示升序，-1 表示降序
 	}
-	err = Unique(indexModelUnique, constant.CNamePlayer)
+	err := Index(indexModelUnique, constant.CNamePlayer)
 	if err != nil {
 		return err
 	}
@@ -120,17 +97,6 @@ func SetPlayerDailyRecordIndex() error {
 }
 
 func SetCNameAccountIndex() error {
-	keys := bson.D{
-		{"userName", 1},
-	}
-	indexModel := mongo.IndexModel{
-		Keys:    keys,
-		Options: options.Index().SetUnique(true), // 设置索引选项，例如唯一索引
-	}
-	err := Unique(indexModel, constant.CNameAccount)
-	if err != nil {
-		return err
-	}
 	keysUnique := bson.D{
 		{"openId", 1},
 	}
@@ -138,7 +104,7 @@ func SetCNameAccountIndex() error {
 		Keys:    keysUnique,                      // 1 表示升序，-1 表示降序
 		Options: options.Index().SetUnique(true), // 设置索引选项，例如唯一索引
 	}
-	err = Unique(indexModelUnique, constant.CNameAccount)
+	err := Unique(indexModelUnique, constant.CNameAccount)
 	if err != nil {
 		return err
 	}

game/game_cluster/nodes/adminapi/web.go

@@ -26,7 +26,7 @@ func Run(profileFilePath, nodeId string) {
 	// 注册db组件
 	app.Register(mhayaMongo.NewComponent())
 	app.AddActors(
-		&mdb.ActorDB{},
+		&mdb.ActorDB{Name: nodeId},
 	)
 	app.Startup()
 }

game/game_cluster/nodes/adminapi/web/web.go

@@ -29,7 +29,7 @@ func main() {
 	app.Register(mhayaMongo.NewComponent())
 
 	app.AddActors(
-		&mdb.ActorDB{},
+		&mdb.ActorDB{Name: "m-web-admin-api-1"},
 	)
 	// 启动mhaya引擎
 	app.Startup()

game/game_cluster/nodes/center/center.go

@@ -26,7 +26,7 @@ func main() {
 	app.AddActors(
 		&account.ActorAccount{},
 		&ops.ActorOps{},
-		&mdb.ActorDB{},
+		&mdb.ActorDB{Name: "m-center"},
 	)
 
 	app.Startup()

game/game_cluster/nodes/center/center/center.go

@@ -26,7 +26,7 @@ func Run(profileFilePath, nodeId string) {
 	app.AddActors(
 		&account.ActorAccount{},
 		&ops.ActorOps{},
-		&mdb.ActorDB{},
+		&mdb.ActorDB{Name: nodeId},
 	)
 
 	app.Startup()

game/game_cluster/nodes/game/module/player/actor_player.go

@@ -7,6 +7,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+	"github.com/go-redis/redis/v8"
 	jsoniter "github.com/json-iterator/go"
 	mhayaString "github.com/mhaya/extend/string"
 	mhayaTime "github.com/mhaya/extend/time"
@@ -43,11 +44,15 @@ type (
 		Player     *models.Player
 		Account    *models.Account
 	}
+
+	CachePlayer struct {
+		Player *models.Player
+		Dirty  bool
+	}
 )
 
 func (p *actorPlayer) OnInit() {
-
-	p.exitTime = time.Minute * 60
+	p.exitTime = time.Minute * 10
 	clog.Debugf("[actorPlayer] path = %s init!", p.PathString())
 
 	// 注册 session关闭的remote函数(网关触发连接断开后，会调用RPC发送该消息)
@@ -98,7 +103,7 @@ func (p *actorPlayer) OnInit() {
 	//p.Timer().Add(10*time.Second, p.refreshRoll)
 	p.Timer().Add(60*time.Second, p.SetRewardToDB)
 	p.Timer().Add(1*time.Minute, p.setRedis)
-	p.Timer().Add(60*time.Second, p.setDb)
+	//p.Timer().Add(60*time.Second, p.setDb)
 	//p.Timer().Add(30*time.Second, p.setInviteReward)
 
 	//p.Timer().Add(1*time.Second, p.addRoll)
@@ -109,6 +114,7 @@ func (p *actorPlayer) OnInit() {
 
 	//在线用户定时刷新
 	p.Timer().Add(1*time.Minute, p.initDaily)
+	p.Timer().Add(10*time.Second, p.setRedisAll)
 
 	//p.Timer().Add(5*time.Second, p.initHourly)
 
@@ -222,6 +228,20 @@ func (p *actorPlayer) SetRewardToDB() {
 	}()
 }
 
+func (p *actorPlayer) setRedisAll() {
+	if p.isOnline && p.Player != nil && p.dirty {
+		ctx := context.Background()
+		cachePlayer := CachePlayer{Dirty: true, Player: p.Player}
+		// 将玩家信息转换为 JSON 格式并存储到 Redis
+		jsonData, err := jsoniter.Marshal(&cachePlayer)
+		if err != nil {
+			return
+		}
+		mdb.RDB.HSet(ctx, fmt.Sprintf("%v:%v", constant.CNamePlayer, p.App().NodeId()), p.Player.OpenId, jsonData)
+		p.dirty = false
+	}
+}
+
 func (p *actorPlayer) setDb() {
 	go func() {
 		if p.isOnline && p.dirty {
@@ -409,6 +429,8 @@ func (p *actorPlayer) sessionClose() {
 	online.UnBindPlayer(p.uid)
 	p.setDb()
 	p.SetRewardToDB()
+	mdb.RDB.HDel(context.Background(), fmt.Sprintf("%v:%v", constant.CNamePlayer, p.App().NodeId()), string(p.uid))
+
 	logoutEvent := event.NewPlayerLogout(p.ActorID(), p.playerId, p.targetPath)
 	p.isOnline = false
 	p.Exit()
@@ -417,18 +439,49 @@ func (p *actorPlayer) sessionClose() {
 
 func (p *actorPlayer) Init(req *token.Token) {
 	clog.Debugf("[actorPlayer] init token = %s", req)
+
+	ctx := context.Background()
+	acc, err := mdb.RDB.Get(ctx, fmt.Sprintf("%v:%v", constant.CNameAccount, req.OpenID)).Bytes()
+	if err != nil && err != redis.Nil {
+		clog.Debugf("[actorPlayer] initUser error: %v", err)
+		return
+	}
 	var account models.Account
-	mdb.MDB.Collection(constant.CNameAccount).FindOne(context.Background(), bson.M{"userName": req.PlayerID}).Decode(&account)
-	p.Account = &account
-	var player models.Player
-	mdb.MDB.Collection(constant.CNamePlayer).FindOne(context.Background(), bson.M{"userName": req.PlayerID}).Decode(&player)
+	if len(acc) > 0 {
+		err = jsoniter.Unmarshal(acc, &account)
+		if err != nil {
+			clog.Debugf("[actorPlayer] unmarshal account data error: %v", err)
+			return
+		}
+		p.Account = &account
+	} else {
+		mdb.MDB.Collection(constant.CNameAccount).FindOne(context.Background(), bson.M{"openId": req.OpenID}).Decode(&account)
+		p.Account = &account
+	}
 
-	if player.OpenId != "" {
-		p.Player = &player
-		clog.Debugf("[Init] oldUser userName = %s openId = %s", p.Player.UserName, p.Player.OpenId)
+	var player models.Player
+	var cachePlayer CachePlayer
+	play, err := mdb.RDB.HGet(ctx, fmt.Sprintf("%v:%v", constant.CNamePlayer, p.App().NodeId()), req.OpenID).Bytes()
+	if err != nil && err != redis.Nil {
+		clog.Debugf("[actorPlayer] initUser error: %v", err)
+		return
+	}
+	if len(play) > 0 {
+		err = jsoniter.Unmarshal(play, &cachePlayer)
+		if err != nil {
+			clog.Debugf("[actorPlayer] unmarshal player data error: %v", err)
+			return
+		}
+		p.Player = cachePlayer.Player
 	} else {
+		mdb.MDB.Collection(constant.CNamePlayer).FindOne(context.Background(), bson.M{"userName": req.PlayerID}).Decode(&player)
+		cachePlayer.Player = &player
+	}
+	if cachePlayer.Player.OpenId == "" {
 		user := models.NewPlayer(req)
-		mdb.MDB.Collection(constant.CNamePlayer).InsertOne(context.Background(), user)
+		//	go mdb.MDB.Collection(constant.CNamePlayer).InsertOne(context.Background(), user)
+		//标记新用户
+		mdb.RDB.HSet(ctx, constant.NewPlayers, req.OpenID, 1)
 		p.Player = user
 		clog.Debugf("[Init] newUser userName = %s openId = %s", p.Player.UserName, p.Player.OpenId)
 		p.botInvite()

game/game_cluster/nodes/game/module/player/actor_players.go

@@ -1,14 +1,22 @@
 package player
 
 import (
+	"context"
+	"fmt"
+	jsoniter "github.com/json-iterator/go"
 	cconst "github.com/mhaya/const"
 	cstring "github.com/mhaya/extend/string"
 	cfacade "github.com/mhaya/facade"
+	"github.com/mhaya/game/game_cluster/internal/constant"
 	"github.com/mhaya/game/game_cluster/internal/event"
+	"github.com/mhaya/game/game_cluster/internal/mdb"
 	"github.com/mhaya/game/game_cluster/internal/mdb/models"
 	"github.com/mhaya/game/game_cluster/nodes/game/module/online"
 	clog "github.com/mhaya/logger"
 	"github.com/mhaya/net/parser/pomelo"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"log"
 	"time"
 )
 
@@ -26,7 +34,7 @@ func (p *ActorPlayers) AliasID() string {
 }
 
 func (p *ActorPlayers) OnInit() {
-	p.childExitTime = time.Minute * 30
+	p.childExitTime = time.Minute * 10
 	// 注册角色登陆事件
 	p.Event().Register(event.PlayerLoginKey, p.onLoginEvent)
 	p.Event().Register(event.PlayerLogoutKey, p.onLogoutEvent)
@@ -34,11 +42,61 @@ func (p *ActorPlayers) OnInit() {
 	p.Remote().Register("checkChild", p.checkChild)
 
 	p.Timer().AddOnce(20*time.Second, p.loadServer)
+	p.Timer().Add(2*time.Minute, p.SetDb)
 }
 
 func (p *ActorPlayers) loadServer() {
 	//初始节点
 	models.SetServerRecord(p.App().NodeId(), 0)
+	mdb.RDB.Del(context.Background(), fmt.Sprintf("%v:%v", constant.CNamePlayer, p.App().NodeId()))
+}
+
+func (p *ActorPlayers) SetDb() {
+	go p.OpDb()
+}
+
+func (p *ActorPlayers) OpDb() {
+	res := mdb.RDB.HGetAll(context.Background(), fmt.Sprintf("%v:%v", constant.CNamePlayer, p.App().NodeId())).Val()
+	newPlayer := mdb.RDB.HGetAll(context.Background(), constant.NewPlayers).Val()
+	bulkUpdOps := []mongo.WriteModel{}
+	bulkInsOps := []mongo.WriteModel{}
+	np := []string{}
+	for k, v := range res {
+		var player CachePlayer
+		if len(v) > 0 {
+			err := jsoniter.Unmarshal([]byte(v), &player)
+			if err != nil {
+				clog.Debugf("[ActorPlayers] unmarshal player data error: %v", err)
+				return
+			}
+			if player.Dirty {
+				if _, ok := newPlayer[k]; ok {
+					np = append(np, k)
+					bulkInsOps = append(bulkInsOps, mongo.NewInsertOneModel().SetDocument(player.Player))
+				} else {
+					filter := bson.M{"openId": player.Player.OpenId}
+					update := bson.M{"$set": player.Player}
+					bulkUpdOps = append(bulkUpdOps, mongo.NewUpdateOneModel().SetFilter(filter).SetUpdate(update))
+				}
+			}
+		}
+	}
+	if len(bulkInsOps) > 0 {
+		_, err := mdb.MDB.Collection(constant.CNamePlayer).BulkWrite(context.Background(), bulkInsOps)
+		if err != nil {
+			log.Fatalf("Bulk write failed: %v", err)
+		}
+		for _, op := range np {
+			mdb.RDB.HDel(context.Background(), constant.NewPlayers, op)
+		}
+
+	}
+	if len(bulkUpdOps) > 0 {
+		_, err := mdb.MDB.Collection(constant.CNamePlayer).BulkWrite(context.Background(), bulkUpdOps)
+		if err != nil {
+			clog.Fatalf("Bulk write failed: %v", err)
+		}
+	}
 }
 
 func (p *ActorPlayers) OnFindChild(msg *cfacade.Message) (cfacade.IActor, bool) {
@@ -130,5 +188,6 @@ func (p *ActorPlayers) onPlayerCreateEvent(e cfacade.IEventData) {
 }
 
 func (p *ActorPlayers) OnStop() {
+	p.SetDb()
 	clog.Infof("onlineCount = %d", online.Count())
 }

game/game_cluster/robot_client/main.go

@@ -17,9 +17,9 @@ import (
 func main() {
 
 	// 定义命令行标志
-	userOffset := flag.Int("userOffset", 0, "用户偏移量")
-	count := flag.Int("count", 80000, "请求总数")
-	qps := flag.Int("qps", 500, "最大并发数")
+	userOffset := flag.Int("userOffset", 300001, "用户偏移量")
+	count := flag.Int("count", 50000, "请求总数")
+	qps := flag.Int("qps", 200, "最大并发数")
 	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")
@@ -55,7 +55,7 @@ func main() {
 					Platform: "ios",
 					Channel:  "tg",
 				}
-				n := rand.Intn(1)
+				n := rand.Intn(4)
 				switch n {
 				case 0:
 					RunRobot(*url, account, i)