package sessions import ( "context" redisComponent "f1-game/internal/component/redis" nameActor "f1-game/internal/name/actor" nameRemote "f1-game/internal/name/remote" "f1-game/internal/pb" "fmt" "sync/atomic" "time" cstring "github.com/cherry-game/cherry/extend/string" clog "github.com/cherry-game/cherry/logger" "github.com/cherry-game/cherry/net/parser/pomelo" cproto "github.com/cherry-game/cherry/net/proto" cprofile "github.com/cherry-game/cherry/profile" "github.com/go-redis/redis/v8" "github.com/goburrow/cache" ) var ( actor = &Actor{} ) type ( Actor struct { pomelo.ActorBase isSync bool // 是否实时同步标识 cache cache.Cache // session cache. key:playeID, value:*cproto.Session count int64 // session count redisClient *redis.Client // redis client ctx context.Context // context redisFields []string // redis查询的字段 redisKeyFormat string // redis key onbindFuncList []OnBindFunc // 绑定触发函数 onUnbindFuncList []OnUnbindFunc // 解绑触发函数 } OnBindFunc func(session *cproto.Session) bool OnUnbindFunc func(playerID int64) ) // isSync = true 则实时同步session到本地进程 func NewActor(isSync bool) *Actor { actor.isSync = isSync actor.ctx = context.Background() actor.redisFields = []string{ UID, SID, AgentPath, } actor.redisKeyFormat = fmt.Sprintf("%s-game:session:", cprofile.Env()) actor.cache = buildCache(isSync, 100000, 1*time.Second) actor.count = 0 return actor } func (p *Actor) AliasID() string { return nameActor.GateSessions } func (p *Actor) OnInit() { if p.isSync { p.Remote().Register(nameRemote.Sessions_Bind, p.bind) p.Remote().Register(nameRemote.Sessions_Unbind, p.ubind) } // 初始化redis p.initRedis() clog.Infof("[sessions] actor init success. isSync = %v", p.isSync) } func (p *Actor) bind(session *cproto.Session) { playerID := GetPlayerID(session) if playerID > 0 { p.addSession2Cache(playerID, session) for _, fn := range p.onbindFuncList { fn(session) } } } func (p *Actor) ubind(req *pb.I64) { if req.Value > 0 { p.delSessoin2Cache(req.Value) for _, fn := range p.onUnbindFuncList { fn(req.Value) } } } func (p *Actor) initRedis() { config := cprofile.GetConfig("sessions") if config.LastError() != nil { clog.Fatal("sessions property not exists.") return } redisID := config.GetString("redis_id") if redisID == "" { clog.Fatal("sessions redis_id property not exists.") return } redisConfigList := cprofile.GetConfig("redis") if redisConfigList.LastError() != nil { clog.Fatal("redis property not exists.") return } redisConfig := redisComponent.GetConfigWithRedisID(redisConfigList, redisID) if redisConfig == nil { clog.Fatalf("redis config not found. [redisID = %s]", redisID) return } _, options := redisComponent.NewClientOption(redisConfig) p.redisClient = redis.NewClient(options) } func (p *Actor) BuildPlayerKey(playerID int64) string { return fmt.Sprintf("%s%d", p.redisKeyFormat, playerID) } func (p *Actor) GetCacheSession(playerID int64) (*cproto.Session, bool) { value, found := p.cache.GetIfPresent(playerID) if !found { return nil, false } return value.(*cproto.Session), true } func (p *Actor) GetCacheSessions(playerIDS []int64) (map[int64]*cproto.Session, []int64) { var ( sessionMaps = map[int64]*cproto.Session{} missPlayerIDS []int64 ) for _, playerID := range playerIDS { if session, found := p.GetCacheSession(playerID); found { sessionMaps[playerID] = session } else { if playerID > 0 { missPlayerIDS = append(missPlayerIDS, playerID) } } } return sessionMaps, missPlayerIDS } func (p *Actor) GetRedisSession(playerID int64) (*cproto.Session, bool) { key := p.BuildPlayerKey(playerID) result, err := p.redisClient.HMGet(p.ctx, key, p.redisFields...).Result() if err != nil { clog.Errorf("[GetRedisSession] error: %v", err) return nil, false } session := result2Session(result) if len(session.AgentPath) > 0 { p.addSession2Cache(playerID, session) // add session to cache return session, true } return nil, false } func (p *Actor) GetRedisSessions(playerIDS []int64) map[int64]*cproto.Session { pipeline := p.redisClient.Pipeline() defer pipeline.Close() cmds := make([]*redis.SliceCmd, 0, len(playerIDS)) for _, playerID := range playerIDS { key := p.BuildPlayerKey(playerID) cmds = append(cmds, pipeline.HMGet(p.ctx, key, p.redisFields...)) } if _, err := pipeline.Exec(p.ctx); err != nil { clog.Errorf("[GetRedisSessions] error: %v", err) return nil } sessionMap := map[int64]*cproto.Session{} for i, playerID := range playerIDS { result, err := cmds[i].Result() if err != nil { continue } session := result2Session(result) if len(session.AgentPath) > 0 && session.Uid > 0 { p.addSession2Cache(playerID, session) // add session to cache sessionMap[playerID] = session } } return sessionMap } // 设置session绑定时触发的函数 func SetOnBindFunc(fn OnBindFunc) { if fn != nil { actor.onbindFuncList = append(actor.onbindFuncList, fn) } } // 设置session解绑时触发的函数 func SetOnUnbindFunc(fn OnUnbindFunc) { if fn != nil { actor.onUnbindFuncList = append(actor.onUnbindFuncList, fn) } } func buildCache(isSync bool, maxSize int, expire time.Duration) cache.Cache { var options []cache.Option if !isSync { // 非实时同步模式,缓存后x秒过期 options = append(options, cache.WithExpireAfterWrite(expire)) } // 缓存最多10万个session options = append(options, cache.WithMaximumSize(maxSize)) // 缓存失效时触发监听函数 options = append(options, cache.WithRemovalListener(func(key cache.Key, _ cache.Value) { // playerID := key.(int64) // session数量-1 atomic.AddInt64(&actor.count, -1) })) return cache.New(options...) } func result2Session(result []interface{}) *cproto.Session { uid := cstring.ToInt64D(cstring.ToString(result[0]), 0) sid := cstring.ToString(result[1]) agentPath := cstring.ToString(result[2]) return &cproto.Session{ Uid: uid, Sid: sid, AgentPath: agentPath, } } func (p *Actor) addSession2Cache(playerID int64, session *cproto.Session) { p.cache.Put(playerID, session) atomic.AddInt64(&p.count, 1) } func (p *Actor) delSessoin2Cache(playerID int64) { p.cache.Invalidate(playerID) }