Compare commits

..

9 Commits

Author SHA1 Message Date
ayflying
33c8712c72 去掉更新日志 2025-09-03 13:46:39 +08:00
ayflying
26763c04e3 修改act持久化通道安全 2025-09-03 12:19:35 +08:00
ayflying
8c60a1f6c7 执行通道进入协程,预防拥堵 2025-09-03 11:14:32 +08:00
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
ayflying
aa1dc0896d 修复计划任务打断造成协程内上下文执行停止的问题 2025-09-02 12:25:30 +08:00
6 changed files with 367 additions and 43 deletions

View File

@@ -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

View File

@@ -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,7 +112,14 @@ func (s *sGameAct) Set(uid int64, actId int, data interface{}) (err error) {
return
}
func (s *sGameAct) Saves(ctx context.Context) (err error) {
// 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数据保存了")
//如果没有执行过,设置时间戳
// 最大允许执行时间
@@ -118,28 +128,29 @@ func (s *sGameAct) Saves(ctx context.Context) (err error) {
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) {
//判断是否超时
@@ -156,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
}
@@ -184,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()
@@ -206,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()
@@ -223,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)
@@ -233,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()
@@ -253,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)
@@ -270,8 +283,272 @@ 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 int64
//批量更新数据库计数
var updateCount int64
//通道关闭标志
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
//}
row, _ := addRes.RowsAffected()
addCount += row
//删除缓存
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 {
@@ -284,12 +561,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

View File

@@ -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,7 +39,8 @@ 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数据保存")
@@ -72,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 {
@@ -94,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,
@@ -103,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
})
@@ -125,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 {

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,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)

View File

@@ -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)
}
)

View File

@@ -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 {