redis刷新列表，允许打断，防止过长的日志

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,12 +17,12 @@ 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
-	})
+	}, true)
 
 	//初始化自启动方法
 	for _, v := range _func {

internal/logic/gameAct/gameAct.go

@@ -2,6 +2,7 @@ package gameAct
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -20,9 +21,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 {
@@ -109,16 +111,24 @@ func (s *sGameAct) Set(uid int64, actId int, data interface{}) (err error) {
 
 func (s *sGameAct) Saves(ctx context.Context) (err error) {
 	getCache, _ := pkg.Cache("redis").Get(nil, "cron:game_act")
+	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 {
-		err = s.Save(ctx, i.(int))
+		//在时间内允许执行
+		if gtime.Now().Before(RunTimeMax) {
+			g.Log().Debug(ctx, "开始执行游戏act数据保存: act%v", i)
+			err = s.Save(ctx, i.(int))
+		} else {
+			g.Log().Errorf(ctx, "游戏act数据保存超时: act=%v", i)
+		}
 		return true
 	})
 	return
@@ -135,6 +145,12 @@ 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, "act执行超时了,停止执行！")
+			err = errors.New("act执行超时了,停止执行！")
+			return
+		}
 		var add = make([]*entity.GameAct, 0)
 		var update = make([]*entity.GameAct, 0)
 		var delKey []string
@@ -195,19 +211,26 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
 		}
 
 		//批量写入数据库
+		updateCount := 0
 		if len(delKey) > 0 {
 			for _, v := range update {
 				v.UpdatedAt = gtime.Now()
-				_, err2 := g.Model(Name).Where(do.GameAct{
+				updateRes, err2 := g.Model(Name).Where(do.GameAct{
-					Uid:       v.Uid,
+					Uid:   v.Uid,
-					ActId:     v.ActId,
+					ActId: v.ActId,
-					UpdatedAt: v.UpdatedAt,
 				}).Data(v).Update()
 				if err2 != nil {
 					g.Log().Error(ctx, err2)
 					return
 				}
+				if row, _ := updateRes.RowsAffected(); row == 0 {
+					g.Log().Error(ctx, "本次更新为0，更新数据失败: %v", v)
+					return
+				}
+				updateCount++
 			}
+			g.Log().Debugf(ctx, "当前 %v 更新数据库: %v 条", actId, updateCount)
+
 			update = make([]*entity.GameAct, 0)
 			var count int64
 
@@ -220,6 +243,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
 					return
 				}
 				count, _ = dbRes.RowsAffected()
+				g.Log().Debugf(ctx, "当前 %v 写入数据库: %v 条", actId, count)
 			}
 
 			for _, v := range delKey {
@@ -230,7 +254,6 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
 			}
 			delKey = make([]string, 0)
 
-			g.Log().Debugf(ctx, "当前 %v 写入数据库: %v 条", actId, count)
 		}
 
 		if err != nil {

internal/logic/gameKv/gameKv.go

@@ -1,6 +1,7 @@
 package gameKv
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -16,8 +17,9 @@ import (
 )
 
 var (
-	ctx  = gctx.New()
+	ctx        = gctx.New()
-	Name = "game_kv"
+	Name       = "game_kv"
+	RunTimeMax *gtime.Time
 )
 
 type sGameKv struct {
@@ -38,6 +40,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 +55,13 @@ 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().Error(ctx, "kv执行超时了,停止执行！")
+			err = errors.New("kv执行超时了,停止执行！")
+			return
+		}
+
 		// 定义用于存储用户数据的结构体
 		type ListData struct {
 			Uid int64       `json:"uid"`
@@ -122,17 +135,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
 }

internal/logic/systemCron/systemCron.go

@@ -2,6 +2,9 @@ package systemCron
 
 import (
 	"context"
+	"sync"
+	"time"
+
 	"github.com/ayflying/utility_go/api/system/v1"
 	"github.com/ayflying/utility_go/service"
 	"github.com/gogf/gf/v2/frame/g"
@@ -9,8 +12,6 @@ import (
 	"github.com/gogf/gf/v2/os/gctx"
 	"github.com/gogf/gf/v2/os/gtime"
 	"github.com/gogf/gf/v2/os/gtimer"
-	"sync"
-	"time"
 )
 
 var (
@@ -80,7 +81,8 @@ func (s *sSystemCron) AddCron(typ v1.CronType, _func func() error) {
 	var _func2 = func(ctx context.Context) error {
 		return _func()
 	}
-	s.AddCronV2(typ, _func2)
+	// 老版本计划任务全都是主服务器唯一执行
+	s.AddCronV2(typ, _func2, true)
 }
 
 // AddCronV2  添加一个定时任务到相应的调度列表中。
@@ -89,7 +91,16 @@ func (s *sSystemCron) AddCron(typ v1.CronType, _func func() error) {
 // @receiver s: sSystemCron的实例，代表一个调度系统。
 // @param typ: 任务的类型，决定该任务将被添加到哪个列表中。对应不同的时间间隔。
 // @param _func: 要添加的任务函数，该函数执行时应该返回一个error。
-func (s *sSystemCron) AddCronV2(typ v1.CronType, _func func(context.Context) error) {
+// @param _onlyMain: 是否只在主服务器上执行一次,true 唯一执行，false 全局执行不判断唯一
+func (s *sSystemCron) AddCronV2(typ v1.CronType, _func func(context.Context) error, _onlyMain ...bool) {
+	//如果传过来的任务是需要主服务器执行一次
+	if len(_onlyMain) > 0 && _onlyMain[0] {
+		//判断当前是否为主服务器
+		if !g.Cfg().MustGet(gctx.New(), "game.cron_main").Bool() {
+			return
+		}
+	}
+
 	//加锁
 	s.Lock.Lock()
 	defer s.Lock.Unlock()

pkg/aycache/cache.go

@@ -1,6 +1,7 @@
 package aycache
 
 import (
+	"context"
 	"math"
 
 	v1 "github.com/ayflying/utility_go/api/system/v1"
@@ -33,7 +34,7 @@ var QPS = promauto.NewGauge(
 func init() {
 	boot.AddFunc(func() {
 		// 初始化指标，每分钟计算一次平均 QPS 并重置计数器
-		service.SystemCron().AddCron(v1.CronType_MINUTE, func() error {
+		service.SystemCron().AddCronV2(v1.CronType_MINUTE, func(context.Context) error {
 			QPS.Set(math.Round(float64(QPSCount) / 60))
 			QPSCount = 0
 			return nil

service/ip_2_region.go

@@ -7,9 +7,6 @@ package service
 
 type (
 	IIp2Region interface {
-		// Load 加载到内存中
-		//
-		//	@Description: 加载ip2region数据库到内存中。
 		//	@receiver s *sIp2region: sIp2region的实例。
 		Load()
 		GetIp(ip string) (res []string)

service/system_cron.go

@@ -38,7 +38,8 @@ type (
 		// @receiver s: sSystemCron的实例，代表一个调度系统。
 		// @param typ: 任务的类型，决定该任务将被添加到哪个列表中。对应不同的时间间隔。
 		// @param _func: 要添加的任务函数，该函数执行时应该返回一个error。
-		AddCronV2(typ v1.CronType, _func func(context.Context) error)
+		// @param unique: 是否只在唯一服务器上执行
+		AddCronV2(typ v1.CronType, _func func(context.Context) error, unique ...bool)
 		// StartCron 开始计划任务执行
 		//
 		//	@Description:

tools/redis.go

@@ -70,6 +70,9 @@ func (r *redis) RedisScanV2(cacheKey string, _func func([]string) error, _key ..
 
 		if len(newKeys) > 0 {
 			err = _func(newKeys)
+			if err != nil {
+				return err
+			}
 		}
 
 		//这个要放在最后

utility.go

@@ -1,11 +1,15 @@
 package utility_go
 
 import (
+	"context"
+	"time"
+
 	"github.com/ayflying/utility_go/config"
 	"github.com/ayflying/utility_go/internal/boot"
 	_ "github.com/ayflying/utility_go/internal/logic"
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/os/gctx"
+	"github.com/gogf/gf/v2/os/gtimer"
 )
 
 var (
@@ -14,9 +18,12 @@ var (
 )
 
 func init() {
+	var err error
 	g.Log().Debug(ctx, "utility_go init启动完成")
 	// 初始化配置
-	var err = boot.Boot()
+	gtimer.SetTimeout(ctx, time.Second*5, func(ctx context.Context) {
+		err = boot.Boot()
+	})
 
 	if err != nil {
 		panic(err)