Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
26763c04e3 | ||
|
|
8c60a1f6c7 | ||
|
|
62b0e429b3 | ||
|
|
e1f1bea0e7 | ||
|
|
299ba0b93e | ||
|
|
c12c49477c | ||
|
|
b052754a30 | ||
|
|
aa1dc0896d | ||
|
|
8210ac24db |
@@ -19,12 +19,12 @@ func Boot() (err error) {
|
||||
//err = service.SystemCron().StartCron()
|
||||
|
||||
//用户活动持久化每小时执行一次
|
||||
service.SystemCron().AddCronV2(v1.CronType_HOUR, func(ctx context.Context) error {
|
||||
service.SystemCron().AddCronV2(v1.CronType_HOUR, func(context.Context) error {
|
||||
go func() {
|
||||
err = service.GameKv().SavesV1(ctx)
|
||||
err = service.GameAct().Saves(ctx)
|
||||
err = service.GameKv().SavesV1()
|
||||
err = service.GameAct().SavesV2()
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, err)
|
||||
g.Log().Error(gctx.New(), err)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/ayflying/utility_go/internal/model/do"
|
||||
@@ -14,6 +15,7 @@ import (
|
||||
service2 "github.com/ayflying/utility_go/service"
|
||||
"github.com/ayflying/utility_go/tools"
|
||||
"github.com/gogf/gf/v2/container/gset"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
"github.com/gogf/gf/v2/os/gtime"
|
||||
@@ -21,7 +23,6 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
ctx = gctx.New()
|
||||
Name = "game_act"
|
||||
ActList = gset.New(true)
|
||||
RunTimeMax *gtime.Time
|
||||
@@ -47,6 +48,7 @@ func init() {
|
||||
// @return data *v1.Act: 返回活动信息结构体指针
|
||||
// @return err error: 返回错误信息
|
||||
func (s *sGameAct) Info(uid int64, actId int) (data *g.Var, err error) {
|
||||
var ctx = gctx.New()
|
||||
if uid == 0 || actId == 0 {
|
||||
g.Log().Error(ctx, "当前参数为空")
|
||||
return
|
||||
@@ -89,6 +91,7 @@ func (s *sGameAct) Info(uid int64, actId int) (data *g.Var, err error) {
|
||||
// @param data interface{}: 要存储的活动信息数据。
|
||||
// @return err error: 返回错误信息,如果操作成功,则返回nil。
|
||||
func (s *sGameAct) Set(uid int64, actId int, data interface{}) (err error) {
|
||||
var ctx = gctx.New()
|
||||
if uid == 0 || actId == 0 {
|
||||
g.Log().Error(ctx, "当前参数为空")
|
||||
return
|
||||
@@ -109,43 +112,45 @@ func (s *sGameAct) Set(uid int64, actId int, data interface{}) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (s *sGameAct) Saves(ctx context.Context) (err error) {
|
||||
getCache, _ := pkg.Cache("redis").Get(nil, "cron:game_act")
|
||||
// Saves 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @return err error: 返回错误信息
|
||||
// Deprecated: 该方法已被弃用,建议使用SavesV2方法
|
||||
func (s *sGameAct) Saves() (err error) {
|
||||
var ctx = gctx.New()
|
||||
g.Log().Debug(ctx, "开始执行游戏act数据保存了")
|
||||
//如果没有执行过,设置时间戳
|
||||
if getCache.Int64() > 0 {
|
||||
return
|
||||
} else {
|
||||
pkg.Cache("redis").Set(nil, "cron:game_act", gtime.Now().Unix(), time.Hour)
|
||||
}
|
||||
// 最大允许执行时间
|
||||
RunTimeMax = gtime.Now().Add(time.Minute * 30)
|
||||
//遍历执行
|
||||
ActList.Iterator(func(i interface{}) bool {
|
||||
//在时间内允许执行
|
||||
if gtime.Now().Before(RunTimeMax) {
|
||||
g.Log().Debug(ctx, "开始执行游戏act数据保存: act%v", i)
|
||||
g.Log().Debugf(ctx, "开始执行游戏act数据保存:act=%v", i)
|
||||
err = s.Save(ctx, i.(int))
|
||||
} else {
|
||||
g.Log().Errorf(ctx, "游戏act数据保存超时: act=%v", i)
|
||||
g.Log().Errorf(ctx, "游戏act数据保存超时:act=%v", i)
|
||||
}
|
||||
return true
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Save 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
// @param actId int: 活动ID
|
||||
// @return err error: 返回错误信息
|
||||
// deprecated: 该方法已被弃用,建议使用SaveV2方法
|
||||
func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
|
||||
cacheKey := fmt.Sprintf("act:%v:*", actId)
|
||||
//获取当前用户的key值
|
||||
//keys, err := utils.RedisScan(cacheKey)
|
||||
//if len(keys) > 10000 {
|
||||
// keys = keys[:10000]
|
||||
//}
|
||||
|
||||
var add = make([]*entity.GameAct, 0)
|
||||
var update = make([]*entity.GameAct, 0)
|
||||
|
||||
//循环获取缓存数据
|
||||
err = tools.Redis.RedisScanV2(cacheKey, func(keys []string) (err error) {
|
||||
//判断是否超时
|
||||
@@ -162,6 +167,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
uid = gconv.Int64(result[2])
|
||||
//uid, err = strconv.ParseInt(result[2], 10, 64)
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, err)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -190,7 +196,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
ActId: actId,
|
||||
}).Fields("uid,act_id").Scan(&data)
|
||||
if err != nil {
|
||||
g.Log().Debugf(ctx, "当前数据错误: %v", cacheKey)
|
||||
g.Log().Errorf(ctx, "当前数据错误: %v", cacheKey)
|
||||
continue
|
||||
}
|
||||
actionData := cacheGet.String()
|
||||
@@ -212,6 +218,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
//批量写入数据库
|
||||
updateCount := 0
|
||||
|
||||
//g.Log().Debugf(ctx, "当前 %v 要更新的数据: %v 条", actId, len(update))
|
||||
if len(update) > 100 {
|
||||
for _, v := range update {
|
||||
v.UpdatedAt = gtime.Now()
|
||||
@@ -229,7 +236,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
}
|
||||
|
||||
//删除缓存
|
||||
go s.DelCacheKey(v.ActId, v.Uid)
|
||||
go s.DelCacheKey(ctx, v.ActId, v.Uid)
|
||||
|
||||
updateCount++
|
||||
update = make([]*entity.GameAct, 0)
|
||||
@@ -239,7 +246,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
}
|
||||
|
||||
var count int64
|
||||
|
||||
//g.Log().Debugf(ctx, "当前 %v 要添加的数据: %v 条", actId, len(add))
|
||||
if len(add) > 100 {
|
||||
dbRes, err2 := g.Model(Name).Data(add).Save()
|
||||
|
||||
@@ -259,7 +266,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
|
||||
for _, v2 := range add {
|
||||
//删除缓存
|
||||
go s.DelCacheKey(v2.ActId, v2.Uid)
|
||||
go s.DelCacheKey(ctx, v2.ActId, v2.Uid)
|
||||
}
|
||||
|
||||
//g.Log().Debugf(ctx, "当前 %v 写入数据库: %v 条", actId, count)
|
||||
@@ -276,8 +283,271 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
// SavesV2 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @return err error: 返回错误信息
|
||||
// SavesV2 保存游戏活动数据
|
||||
func (s *sGameAct) SavesV2() (err error) {
|
||||
var ctx = gctx.New()
|
||||
g.Log().Debug(ctx, "开始执行游戏act数据保存了")
|
||||
RunTimeMax = gtime.Now().Add(time.Minute * 30)
|
||||
|
||||
// 使用局部通道替代包级通道,避免并发冲突
|
||||
addChan := make(chan *entity.GameAct, 1000)
|
||||
updateChan := make(chan *entity.GameAct, 1000)
|
||||
|
||||
errChan := make(chan error, 1)
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1) // 仅需添加1次(对应Cache2SqlChan协程)
|
||||
// wg.Add(1) // 移除多余的Add调用,避免计数不平衡
|
||||
|
||||
go func() {
|
||||
defer wg.Done() // Cache2SqlChan协程完成后减1
|
||||
s.Cache2SqlChan(ctx, addChan, updateChan)
|
||||
}()
|
||||
|
||||
go func() {
|
||||
scanErr := tools.Redis.RedisScanV2("act:*", func(keys []string) error {
|
||||
if gtime.Now().After(RunTimeMax) {
|
||||
return errors.New("Redis扫描超时")
|
||||
}
|
||||
for _, key := range keys {
|
||||
if keyErr := s.SaveV2(ctx, key, addChan, updateChan); keyErr != nil {
|
||||
g.Log().Errorf(ctx, "处理key %s失败: %v", key, keyErr)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
close(addChan)
|
||||
close(updateChan)
|
||||
errChan <- scanErr
|
||||
}()
|
||||
|
||||
// 等待扫描和处理完成,同时监听上下文取消
|
||||
select {
|
||||
case scanErr := <-errChan:
|
||||
wg.Wait() // 等待Cache2SqlChan处理完剩余数据
|
||||
if scanErr != nil {
|
||||
return gerror.New(fmt.Sprintf("Redis扫描失败: %v", scanErr))
|
||||
}
|
||||
case <-ctx.Done():
|
||||
wg.Wait()
|
||||
return ctx.Err() // 返回上下文取消原因
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// SaveV2 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
// @param cacheKey string: 缓存键
|
||||
// @param add []*entity.GameAct: 添加数据
|
||||
// @param update []*entity.GameAct: 更新数据
|
||||
// @return err error: 返回错误信息
|
||||
func (s *sGameAct) SaveV2(ctx context.Context, cacheKey string, addChan, updateChan chan *entity.GameAct) (err error) {
|
||||
|
||||
result := strings.Split(cacheKey, ":")
|
||||
actId := gconv.Int(result[1])
|
||||
if actId == 0 {
|
||||
return
|
||||
}
|
||||
var uid int64
|
||||
uid = gconv.Int64(result[2])
|
||||
if uid == 0 {
|
||||
//跳过为空的用户缓存
|
||||
return
|
||||
}
|
||||
|
||||
//获取缓存数据
|
||||
cacheGet, _ := g.Redis().Get(ctx, cacheKey)
|
||||
|
||||
if cacheGet.IsEmpty() {
|
||||
//空数据也不保存
|
||||
return
|
||||
}
|
||||
|
||||
//如果有活跃,跳过持久化
|
||||
if getBool, _ := pkg.Cache("redis").
|
||||
Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
|
||||
return
|
||||
}
|
||||
|
||||
//获取数据库数据
|
||||
var data *entity.GameAct
|
||||
// 从数据库中查询活动信息
|
||||
err = g.Model(Name).Where(do.GameAct{
|
||||
Uid: uid,
|
||||
ActId: actId,
|
||||
}).Fields("uid,act_id").Scan(&data)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "当前数据错误: %v", cacheKey)
|
||||
return
|
||||
}
|
||||
|
||||
//如果没有数据,添加
|
||||
actionData := cacheGet.String()
|
||||
if data == nil {
|
||||
//add = append(add, &entity.GameAct{
|
||||
// ActId: actId,
|
||||
// Uid: uid,
|
||||
// Action: actionData,
|
||||
//})
|
||||
addChan <- &entity.GameAct{
|
||||
ActId: actId,
|
||||
Uid: uid,
|
||||
Action: actionData,
|
||||
}
|
||||
} else {
|
||||
//覆盖数据
|
||||
data.ActId = actId
|
||||
data.Uid = uid
|
||||
data.Action = actionData
|
||||
//update = append(update, data)
|
||||
updateChan <- data
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// Cache2Sql 缓存持久化到数据库
|
||||
// @Description: 缓存持久化到数据库
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
// @param add []*entity.GameAct: 添加数据
|
||||
// @param update []*entity.GameAct: 更新数据
|
||||
// @return err error: 返回错误信息
|
||||
func (s *sGameAct) Cache2Sql(ctx context.Context, add, update []*entity.GameAct) {
|
||||
//批量写入数据库
|
||||
updateCount := 0
|
||||
if len(update) > 0 {
|
||||
for _, v := range update {
|
||||
v.UpdatedAt = gtime.Now()
|
||||
updateRes, err2 := g.Model(Name).Where(do.GameAct{
|
||||
Uid: v.Uid,
|
||||
ActId: v.ActId,
|
||||
}).Data(v).Update()
|
||||
if err2 != nil {
|
||||
g.Log().Error(ctx, err2)
|
||||
continue
|
||||
}
|
||||
if row, _ := updateRes.RowsAffected(); row == 0 {
|
||||
g.Log().Error(ctx, "本次更新为0,更新数据失败: %v", v)
|
||||
continue
|
||||
}
|
||||
//删除缓存
|
||||
s.DelCacheKey(ctx, v.ActId, v.Uid)
|
||||
updateCount++
|
||||
}
|
||||
g.Log().Debugf(ctx, "act当前更新数据库: %v 条", updateCount)
|
||||
update = (update)[:0]
|
||||
}
|
||||
|
||||
var addCount int64
|
||||
if len(add) > 0 {
|
||||
for _, v := range add {
|
||||
addRes, err2 := g.Model(Name).Data(v).Insert()
|
||||
if err2 != nil {
|
||||
g.Log().Error(ctx, err2)
|
||||
continue
|
||||
}
|
||||
if row, _ := addRes.RowsAffected(); row == 0 {
|
||||
g.Log().Error(ctx, "本次新增为0,新增数据失败: %v", v)
|
||||
continue
|
||||
}
|
||||
addCount++
|
||||
//删除缓存
|
||||
s.DelCacheKey(ctx, v.ActId, v.Uid)
|
||||
}
|
||||
g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
|
||||
add = (add)[:0]
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Cache2AddChan 批量添加数据库
|
||||
func (s *sGameAct) Cache2SqlChan(ctx context.Context, addChan, updateChan chan *entity.GameAct) {
|
||||
//批量写入数据库计数
|
||||
var addCount int
|
||||
//批量更新数据库计数
|
||||
var updateCount int
|
||||
//通道关闭标志
|
||||
addClosed := false
|
||||
updateClosed := false
|
||||
|
||||
// 使用链式安全模式
|
||||
var db = g.Model(Name).Safe()
|
||||
|
||||
for {
|
||||
//检查是否两个通道都已关闭且为空
|
||||
if addClosed && updateClosed {
|
||||
break
|
||||
}
|
||||
|
||||
select {
|
||||
case v, ok := <-addChan:
|
||||
if !ok {
|
||||
addClosed = true // 仅标记关闭,不立即日志
|
||||
continue
|
||||
}
|
||||
addRes, err2 := db.Data(v).Insert()
|
||||
if err2 != nil {
|
||||
g.Log().Error(ctx, err2)
|
||||
continue
|
||||
}
|
||||
if row, _ := addRes.RowsAffected(); row == 0 {
|
||||
g.Log().Error(ctx, "本次新增为0,新增数据失败: %v", v)
|
||||
continue
|
||||
}
|
||||
addCount++
|
||||
//删除缓存
|
||||
s.DelCacheKey(ctx, v.ActId, v.Uid)
|
||||
|
||||
case v, ok := <-updateChan:
|
||||
if !ok {
|
||||
updateClosed = true // 仅标记关闭,不立即日志
|
||||
continue
|
||||
}
|
||||
v.UpdatedAt = gtime.Now()
|
||||
updateRes, err2 := db.Where(do.GameAct{
|
||||
Uid: v.Uid,
|
||||
ActId: v.ActId,
|
||||
}).Data(v).Update()
|
||||
if err2 != nil {
|
||||
g.Log().Error(ctx, err2)
|
||||
continue
|
||||
}
|
||||
if row, _ := updateRes.RowsAffected(); row == 0 {
|
||||
g.Log().Error(ctx, "本次更新为0,更新数据失败: %v", v)
|
||||
continue
|
||||
}
|
||||
//删除缓存
|
||||
s.DelCacheKey(ctx, v.ActId, v.Uid)
|
||||
updateCount++
|
||||
|
||||
case <-ctx.Done():
|
||||
g.Log().Debug(ctx, "act协程被上下文取消")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// 仅在所有通道处理完毕后打印最终计数(移除中间冗余日志)
|
||||
g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
|
||||
g.Log().Debugf(ctx, "act当前更新数据库: %v 条", updateCount)
|
||||
return
|
||||
}
|
||||
|
||||
// 删除缓存key
|
||||
func (s *sGameAct) DelCacheKey(aid int, uid int64) {
|
||||
func (s *sGameAct) DelCacheKey(ctx context.Context, aid int, uid int64) {
|
||||
//如果有活跃,跳过删除
|
||||
if getBool, _ := pkg.Cache("redis").
|
||||
Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
|
||||
return
|
||||
}
|
||||
|
||||
cacheKey := fmt.Sprintf("act:%v:%v", aid, uid)
|
||||
_, err := g.Redis().Del(ctx, cacheKey)
|
||||
if err != nil {
|
||||
@@ -290,12 +560,13 @@ func (s *sGameAct) RefreshGetRedDotCache(uid int64) {
|
||||
cacheKey := fmt.Sprintf("gameAct:GetRedDot:%s:%d", gtime.Now().Format("d"), uid)
|
||||
_, err := pkg.Cache("redis").Remove(gctx.New(), cacheKey)
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, err)
|
||||
g.Log().Error(gctx.New(), err)
|
||||
g.Dump(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *sGameAct) Del(uid int64, actId int) {
|
||||
var ctx = gctx.New()
|
||||
if uid == 0 || actId == 0 {
|
||||
g.Log().Error(ctx, "当前参数为空")
|
||||
return
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -15,10 +14,10 @@ import (
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
"github.com/gogf/gf/v2/os/gtime"
|
||||
"github.com/gogf/gf/v2/util/gconv"
|
||||
)
|
||||
|
||||
var (
|
||||
ctx = gctx.New()
|
||||
Name = "game_kv"
|
||||
RunTimeMax *gtime.Time
|
||||
)
|
||||
@@ -40,19 +39,12 @@ func init() {
|
||||
// @Description: 保存用户KV数据列表。
|
||||
// @receiver s: sGameKv的实例。
|
||||
// @return err: 错误信息,如果操作成功,则为nil。
|
||||
func (s *sGameKv) SavesV1(ctx context.Context) (err error) {
|
||||
func (s *sGameKv) SavesV1() (err error) {
|
||||
var ctx = gctx.New()
|
||||
// 最大允许执行时间
|
||||
RunTimeMax = gtime.Now().Add(time.Minute * 30)
|
||||
g.Log().Debug(ctx, "开始执行游戏kv数据保存")
|
||||
|
||||
getCache, err := pkg.Cache("redis").Get(nil, "cron:game_kv")
|
||||
//如果没有执行过,设置时间戳
|
||||
if getCache.Int64() > 0 {
|
||||
return
|
||||
} else {
|
||||
pkg.Cache("redis").Set(nil, "cron:game_kv", gtime.Now().Unix(), time.Hour)
|
||||
}
|
||||
|
||||
// 定义用于存储用户数据的结构体
|
||||
type ListData struct {
|
||||
Uid int64 `json:"uid"`
|
||||
@@ -80,19 +72,17 @@ func (s *sGameKv) SavesV1(ctx context.Context) (err error) {
|
||||
//uid := v.Int64()
|
||||
//cacheKey = "user:kv:" + strconv.FormatInt(uid, 10)
|
||||
result := strings.Split(cacheKey, ":")
|
||||
var uid int64
|
||||
uid, err = strconv.ParseInt(result[2], 10, 64)
|
||||
var uid = gconv.Int64(result[2])
|
||||
if uid == 0 {
|
||||
continue
|
||||
}
|
||||
//uid, err = strconv.ParseInt(result[2], 10, 64)
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, err)
|
||||
g.Redis().Del(ctx, cacheKey)
|
||||
continue
|
||||
}
|
||||
|
||||
////如果1天没有活跃,跳过
|
||||
//user, _ := service.MemberUser().Info(uid)
|
||||
//if user.UpdatedAt.Seconds < gtime.Now().Add(consts.ActSaveTime).Unix() {
|
||||
// continue
|
||||
//}
|
||||
//如果有活跃,跳过持久化
|
||||
if getBool, _ := pkg.Cache("redis").
|
||||
Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
|
||||
@@ -102,6 +92,9 @@ func (s *sGameKv) SavesV1(ctx context.Context) (err error) {
|
||||
get, _ := g.Redis().Get(ctx, cacheKey)
|
||||
var data interface{}
|
||||
get.Scan(&data)
|
||||
if data == nil {
|
||||
continue
|
||||
}
|
||||
list = append(list, &ListData{
|
||||
Uid: uid,
|
||||
Kv: data,
|
||||
@@ -111,21 +104,17 @@ func (s *sGameKv) SavesV1(ctx context.Context) (err error) {
|
||||
// 将列表数据保存到数据库
|
||||
if len(list) > 100 {
|
||||
_, err2 := g.Model("game_kv").Data(list).Save()
|
||||
|
||||
if err2 != nil {
|
||||
g.Log().Error(ctx, err2)
|
||||
g.Log().Error(ctx, "当前kv数据入库失败: %v", err2)
|
||||
err = err2
|
||||
return
|
||||
}
|
||||
//删除当前key
|
||||
for _, v := range list {
|
||||
go s.DelCacheKey(v.Uid)
|
||||
s.DelCacheKey(ctx, v.Uid)
|
||||
}
|
||||
list = make([]*ListData, 0)
|
||||
}
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, "当前kv数据入库失败: %v", err)
|
||||
}
|
||||
|
||||
return
|
||||
})
|
||||
|
||||
@@ -133,7 +122,13 @@ func (s *sGameKv) SavesV1(ctx context.Context) (err error) {
|
||||
}
|
||||
|
||||
// 删除缓存key
|
||||
func (s *sGameKv) DelCacheKey(uid int64) {
|
||||
func (s *sGameKv) DelCacheKey(ctx context.Context, uid int64) {
|
||||
//如果有活跃,跳过删除
|
||||
if getBool, _ := pkg.Cache("redis").
|
||||
Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
|
||||
return
|
||||
}
|
||||
|
||||
cacheKey := fmt.Sprintf("user:kv:%v", uid)
|
||||
_, err := g.Redis().Del(ctx, cacheKey)
|
||||
if err != nil {
|
||||
|
||||
@@ -8,6 +8,7 @@ package service
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/ayflying/utility_go/internal/model/entity"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
@@ -31,8 +32,50 @@ type (
|
||||
// @param data interface{}: 要存储的活动信息数据。
|
||||
// @return err error: 返回错误信息,如果操作成功,则返回nil。
|
||||
Set(uid int64, actId int, data interface{}) (err error)
|
||||
Saves(ctx context.Context) (err error)
|
||||
// Saves 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @return err error: 返回错误信息
|
||||
// Deprecated: 该方法已被弃用,建议使用SavesV2方法
|
||||
Saves() (err error)
|
||||
// Save 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
// @param actId int: 活动ID
|
||||
// @return err error: 返回错误信息
|
||||
// deprecated: 该方法已被弃用,建议使用SaveV2方法
|
||||
Save(ctx context.Context, actId int) (err error)
|
||||
// SavesV2 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @return err error: 返回错误信息
|
||||
SavesV2() (err error)
|
||||
// SaveV2 保存游戏活动数据
|
||||
//
|
||||
// @Description: 保存游戏活动数据
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
// @param cacheKey string: 缓存键
|
||||
// @param add []*entity.GameAct: 添加数据
|
||||
// @param update []*entity.GameAct: 更新数据
|
||||
// @return err error: 返回错误信息
|
||||
SaveV2(ctx context.Context, cacheKey string, addChan chan *entity.GameAct, updateChan chan *entity.GameAct) (err error)
|
||||
// Cache2Sql 缓存持久化到数据库
|
||||
// @Description: 缓存持久化到数据库
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
// @param add []*entity.GameAct: 添加数据
|
||||
// @param update []*entity.GameAct: 更新数据
|
||||
// @return err error: 返回错误信息
|
||||
Cache2Sql(ctx context.Context, add []*entity.GameAct, update []*entity.GameAct)
|
||||
// Cache2AddChan 批量添加数据库
|
||||
Cache2SqlChan(ctx context.Context, addChan chan *entity.GameAct, updateChan chan *entity.GameAct)
|
||||
// 删除缓存key
|
||||
DelCacheKey(ctx context.Context, aid int, uid int64)
|
||||
// 清空GetRedDot缓存
|
||||
RefreshGetRedDotCache(uid int64)
|
||||
Del(uid int64, actId int)
|
||||
|
||||
@@ -16,7 +16,9 @@ type (
|
||||
// @Description: 保存用户KV数据列表。
|
||||
// @receiver s: sGameKv的实例。
|
||||
// @return err: 错误信息,如果操作成功,则为nil。
|
||||
SavesV1(ctx context.Context) (err error)
|
||||
SavesV1() (err error)
|
||||
// 删除缓存key
|
||||
DelCacheKey(ctx context.Context, uid int64)
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@@ -48,7 +48,6 @@ func (r *redis) RedisScan(cacheKey string, _key ...string) (keys []string, err e
|
||||
|
||||
// redis 批量获取大量数据
|
||||
func (r *redis) RedisScanV2(cacheKey string, _func func([]string) error, _key ...string) error {
|
||||
|
||||
//var keys []string
|
||||
var err error
|
||||
|
||||
@@ -67,7 +66,6 @@ func (r *redis) RedisScanV2(cacheKey string, _func func([]string) error, _key ..
|
||||
g.Log().Errorf(ctx, "Scan failed: %v", err)
|
||||
break
|
||||
}
|
||||
|
||||
if len(newKeys) > 0 {
|
||||
err = _func(newKeys)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user