Compare commits

...

5 Commits

Author SHA1 Message Date
ayflying
62b0e429b3 如果检测到当前正在活跃,不删除缓存key 2025-09-03 10:02:34 +08:00
ayflying
e1f1bea0e7 增加有优雅的持久化管道,提高上传速度与抗打断机制 2025-09-02 18:16:55 +08:00
ayflying
299ba0b93e 补充漏掉更新统计 2025-09-02 17:08:37 +08:00
ayflying
c12c49477c 优化持久化因为活动太多造成的速度太慢 2025-09-02 16:51:41 +08:00
ayflying
b052754a30 删除多余的日志 2025-09-02 15:12:43 +08:00
4 changed files with 295 additions and 19 deletions

View File

@@ -22,7 +22,7 @@ func Boot() (err error) {
service.SystemCron().AddCronV2(v1.CronType_HOUR, func(context.Context) error {
go func() {
err = service.GameKv().SavesV1()
err = service.GameAct().Saves()
err = service.GameAct().SavesV2()
if err != nil {
g.Log().Error(gctx.New(), err)
}

View File

@@ -24,6 +24,8 @@ var (
Name = "game_act"
ActList = gset.New(true)
RunTimeMax *gtime.Time
addChan chan *entity.GameAct
updateChan chan *entity.GameAct
)
type sGameAct struct {
@@ -110,6 +112,12 @@ func (s *sGameAct) Set(uid int64, actId int, data interface{}) (err error) {
return
}
// 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数据保存了")
@@ -130,12 +138,19 @@ func (s *sGameAct) Saves() (err error) {
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)
var add = make([]*entity.GameAct, 0)
var update = make([]*entity.GameAct, 0)
//循环获取缓存数据
err = tools.Redis.RedisScanV2(cacheKey, func(keys []string) (err error) {
//判断是否超时
@@ -203,7 +218,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
//批量写入数据库
updateCount := 0
g.Log().Debugf(ctx, "当前 %v 要更新的数据: %v 条", actId, len(update))
//g.Log().Debugf(ctx, "当前 %v 要更新的数据: %v 条", actId, len(update))
if len(update) > 100 {
for _, v := range update {
v.UpdatedAt = gtime.Now()
@@ -231,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))
//g.Log().Debugf(ctx, "当前 %v 要添加的数据: %v 条", actId, len(add))
if len(add) > 100 {
dbRes, err2 := g.Model(Name).Data(add).Save()
@@ -268,8 +283,227 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
return
}
// SavesV2 保存游戏活动数据
//
// @Description: 保存游戏活动数据
// @receiver s *sGameAct: 游戏活动服务结构体指针
// @return err error: 返回错误信息
func (s *sGameAct) SavesV2() (err error) {
var ctx = gctx.New()
g.Log().Debug(ctx, "开始执行游戏act数据保存了")
//如果没有执行过,设置时间戳
// 最大允许执行时间
RunTimeMax = gtime.Now().Add(time.Minute * 30)
//cacheKey := fmt.Sprintf("act:%v:*", actId)
addChan = make(chan *entity.GameAct, 1000)
updateChan = make(chan *entity.GameAct, 1000)
go func() {
//循环获取缓存数据
err = tools.Redis.RedisScanV2("act:*", func(keys []string) (err error) {
for _, key := range keys {
//格式化数据
err = s.SaveV2(ctx, key)
}
return err
})
//关闭通道
close(addChan)
close(updateChan)
}()
// 启动缓存数据到数据库通道
s.Cache2SqlChan(ctx)
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) (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
}
// Cache2SqlChan 缓存持久化到数据库
// @Description: 缓存持久化到数据库
// @receiver s *sGameAct: 游戏活动服务结构体指针
// @param ctx context.Context: 上下文对象
func (s *sGameAct) Cache2SqlChan(ctx context.Context) {
//批量写入数据库
updateCount := 0
for v := range updateChan {
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)
var addCount int64
for v := range addChan {
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)
return
}
// 删除缓存key
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 {

View File

@@ -4,7 +4,6 @@ import (
"context"
"errors"
"fmt"
"strconv"
"strings"
"sync"
"time"
@@ -15,6 +14,7 @@ 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 (
@@ -72,19 +72,17 @@ func (s *sGameKv) SavesV1() (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 {
@@ -94,6 +92,9 @@ func (s *sGameKv) SavesV1() (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,
@@ -103,21 +104,17 @@ func (s *sGameKv) SavesV1() (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(ctx, v.Uid)
s.DelCacheKey(ctx, v.Uid)
}
list = make([]*ListData, 0)
}
if err != nil {
g.Log().Error(ctx, "当前kv数据入库失败: %v", err)
}
return
})
@@ -126,6 +123,12 @@ func (s *sGameKv) SavesV1() (err error) {
// 删除缓存key
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 {

View File

@@ -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,46 @@ type (
// @param data interface{}: 要存储的活动信息数据。
// @return err error: 返回错误信息如果操作成功则返回nil。
Set(uid int64, actId int, data interface{}) (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) (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)
// 删除缓存key
DelCacheKey(ctx context.Context, aid int, uid int64)
// 清空GetRedDot缓存