执行持久化的计划任务，有一个打断机制

internal/boot/boot.go

@@ -2,6 +2,7 @@ package boot
 
 import (
 	"context"
+
 	v1 "github.com/ayflying/utility_go/api/system/v1"
 	"github.com/ayflying/utility_go/service"
 	"github.com/gogf/gf/v2/os/gctx"
@@ -16,8 +17,8 @@ func Boot() (err error) {
 	// 启动计划任务定时器，预防debug工具激活计划任务造成重复执行，此处不执行计划任务
 	//err = service.SystemCron().StartCron()
 
-	//用户活动持久化
+	//用户活动持久化每小时执行一次
-	service.SystemCron().AddCronV2(v1.CronType_DAILY, func(ctx context.Context) error {
+	service.SystemCron().AddCronV2(v1.CronType_HOUR, func(ctx context.Context) error {
 		err = service.GameKv().SavesV1()
 		err = service.GameAct().Saves(ctx)
 		return err

internal/logic/gameAct/gameAct.go

@@ -20,9 +20,10 @@ import (
 )
 
 var (
-	ctx     = gctx.New()
+	ctx        = gctx.New()
-	Name    = "game_act"
+	Name       = "game_act"
-	ActList = gset.New(true)
+	ActList    = gset.New(true)
+	RunTimeMax *gtime.Time
 )
 
 type sGameAct struct {
@@ -115,10 +116,15 @@ func (s *sGameAct) Saves(ctx context.Context) (err error) {
 	} 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 {
-		err = s.Save(ctx, i.(int))
+		//在时间内允许执行
+		if gtime.Now().Before(RunTimeMax) {
+			g.Log().Errorf(ctx, "开始执行游戏act数据保存: act%v", i)
+			err = s.Save(ctx, i.(int))
+		}
 		return true
 	})
 	return
@@ -135,6 +141,11 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
 
 	//循环获取缓存数据
 	err = tools.Redis.RedisScanV2(cacheKey, func(keys []string) (err error) {
+		//判断是否超时
+		if gtime.Now().After(RunTimeMax) {
+			g.Log().Debug(ctx, "执行超时了,停止执行！")
+			return
+		}
 		var add = make([]*entity.GameAct, 0)
 		var update = make([]*entity.GameAct, 0)
 		var delKey []string

internal/logic/gameKv/gameKv.go

@@ -16,8 +16,9 @@ import (
 )
 
 var (
-	ctx  = gctx.New()
+	ctx        = gctx.New()
-	Name = "game_kv"
+	Name       = "game_kv"
+	RunTimeMax *gtime.Time
 )
 
 type sGameKv struct {
@@ -38,6 +39,10 @@ func init() {
 // @receiver s: sGameKv的实例。
 // @return err: 错误信息，如果操作成功，则为nil。
 func (s *sGameKv) SavesV1() (err error) {
+	// 最大允许执行时间
+	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 {
@@ -49,6 +54,12 @@ func (s *sGameKv) SavesV1() (err error) {
 	// 从Redis列表中获取所有用户KV索引的键
 	//keys, err := utils.RedisScan("user:kv:*")
 	err = tools.Redis.RedisScanV2("user:kv:*", func(keys []string) (err error) {
+		//判断是否超时
+		if gtime.Now().After(RunTimeMax) {
+			g.Log().Debug(ctx, "执行超时了,停止执行！")
+			return
+		}
+
 		// 定义用于存储用户数据的结构体
 		type ListData struct {
 			Uid int64       `json:"uid"`
@@ -122,17 +133,5 @@ func (s *sGameKv) SavesV1() (err error) {
 		return
 	})
 
-	//if err != nil {
-	//	return err
-	//}
-	////跳过
-	//if len(keys) == 0 {
-	//	return
-	//}
-	////一次最多处理10w条
-	//if len(keys) > 10000 {
-	//	keys = keys[:10000]
-	//}
-
 	return
 }