Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8c60a1f6c7 | ||
|
|
62b0e429b3 | ||
|
|
e1f1bea0e7 | ||
|
|
299ba0b93e | ||
|
|
c12c49477c | ||
|
|
b052754a30 | ||
|
|
aa1dc0896d | ||
|
|
8210ac24db | ||
|
|
cd3de96761 | ||
|
|
ce8ae4d26a | ||
|
|
50cfc23ad2 | ||
|
|
95539038c0 | ||
|
|
4b08a9ce84 | ||
|
|
6efdac7bab | ||
|
|
788cb2e6d4 | ||
|
|
dd8c05b344 |
@@ -5,6 +5,7 @@ import (
|
||||
|
||||
v1 "github.com/ayflying/utility_go/api/system/v1"
|
||||
"github.com/ayflying/utility_go/service"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/gctx"
|
||||
)
|
||||
|
||||
@@ -18,10 +19,15 @@ 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()
|
||||
err = service.GameAct().Saves(ctx)
|
||||
return err
|
||||
err = service.GameAct().SavesV2()
|
||||
if err != nil {
|
||||
g.Log().Error(gctx.New(), err)
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}, true)
|
||||
|
||||
//初始化自启动方法
|
||||
|
||||
@@ -2,6 +2,7 @@ package gameAct
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -20,10 +21,11 @@ import (
|
||||
)
|
||||
|
||||
var (
|
||||
ctx = gctx.New()
|
||||
Name = "game_act"
|
||||
ActList = gset.New(true)
|
||||
RunTimeMax *gtime.Time
|
||||
addChan chan *entity.GameAct
|
||||
updateChan chan *entity.GameAct
|
||||
)
|
||||
|
||||
type sGameAct struct {
|
||||
@@ -46,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
|
||||
@@ -88,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
|
||||
@@ -108,22 +112,23 @@ 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)
|
||||
@@ -133,25 +138,28 @@ func (s *sGameAct) Saves(ctx context.Context) (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)
|
||||
//获取当前用户的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) {
|
||||
//判断是否超时
|
||||
if gtime.Now().After(RunTimeMax) {
|
||||
g.Log().Debug(ctx, "执行超时了,停止执行!")
|
||||
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
|
||||
|
||||
for _, cacheKey = range keys {
|
||||
result := strings.Split(cacheKey, ":")
|
||||
actId, err = strconv.Atoi(result[1])
|
||||
@@ -159,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
|
||||
}
|
||||
|
||||
@@ -187,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()
|
||||
@@ -204,47 +213,64 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
|
||||
data.Action = actionData
|
||||
update = append(update, data)
|
||||
}
|
||||
//最后删除key
|
||||
delKey = append(delKey, cacheKey)
|
||||
}
|
||||
|
||||
//批量写入数据库
|
||||
if len(delKey) > 0 {
|
||||
updateCount := 0
|
||||
|
||||
//g.Log().Debugf(ctx, "当前 %v 要更新的数据: %v 条", actId, len(update))
|
||||
if len(update) > 100 {
|
||||
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,
|
||||
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)
|
||||
continue
|
||||
}
|
||||
update = make([]*entity.GameAct, 0)
|
||||
var count int64
|
||||
|
||||
if len(add) > 0 {
|
||||
dbRes, err2 := g.Model(Name).Batch(50).Data(add).Save()
|
||||
add = make([]*entity.GameAct, 0)
|
||||
//删除缓存
|
||||
go s.DelCacheKey(ctx, v.ActId, v.Uid)
|
||||
|
||||
updateCount++
|
||||
update = make([]*entity.GameAct, 0)
|
||||
}
|
||||
g.Log().Debugf(ctx, "当前 %v 更新数据库: %v 条", actId, updateCount)
|
||||
update = make([]*entity.GameAct, 0)
|
||||
}
|
||||
|
||||
var count int64
|
||||
//g.Log().Debugf(ctx, "当前 %v 要添加的数据: %v 条", actId, len(add))
|
||||
if len(add) > 100 {
|
||||
dbRes, err2 := g.Model(Name).Data(add).Save()
|
||||
|
||||
err = err2
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, err2)
|
||||
return
|
||||
}
|
||||
count, _ = dbRes.RowsAffected()
|
||||
if count == 0 {
|
||||
g.Log().Error(ctx, "当前 %v 写入数据库: %v 条", actId, count)
|
||||
for _, vTemp := range add {
|
||||
g.Log().Debugf(ctx, "当前act:%v,add写入数据: %v,内容:%v", vTemp.ActId, vTemp.Uid, vTemp.Action)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
for _, v := range delKey {
|
||||
_, err = g.Redis().Del(ctx, v)
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, err)
|
||||
for _, v2 := range add {
|
||||
//删除缓存
|
||||
go s.DelCacheKey(ctx, v2.ActId, v2.Uid)
|
||||
}
|
||||
}
|
||||
delKey = make([]string, 0)
|
||||
|
||||
g.Log().Debugf(ctx, "当前 %v 写入数据库: %v 条", actId, count)
|
||||
//g.Log().Debugf(ctx, "当前 %v 写入数据库: %v 条", actId, count)
|
||||
add = make([]*entity.GameAct, 0)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
@@ -257,17 +283,252 @@ 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)
|
||||
}()
|
||||
// 启动缓存数据到数据库通道
|
||||
go s.Cache2AddChan(ctx)
|
||||
go s.Cache2UpdateChan(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
|
||||
}
|
||||
|
||||
// Cache2UpdateChan 缓存持久化到数据库
|
||||
// @Description: 缓存持久化到数据库
|
||||
// @receiver s *sGameAct: 游戏活动服务结构体指针
|
||||
// @param ctx context.Context: 上下文对象
|
||||
func (s *sGameAct) Cache2UpdateChan(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)
|
||||
return
|
||||
}
|
||||
|
||||
// Cache2AddChan 批量添加数据库
|
||||
func (s *sGameAct) Cache2AddChan(ctx context.Context) {
|
||||
//批量写入数据库
|
||||
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 {
|
||||
g.Log().Error(ctx, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 清空GetRedDot缓存
|
||||
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
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
package gameKv
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -13,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
|
||||
)
|
||||
@@ -39,27 +40,11 @@ func init() {
|
||||
// @receiver s: sGameKv的实例。
|
||||
// @return err: 错误信息,如果操作成功,则为nil。
|
||||
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)
|
||||
}
|
||||
|
||||
// 从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, "执行超时了,停止执行!")
|
||||
return
|
||||
}
|
||||
|
||||
// 定义用于存储用户数据的结构体
|
||||
type ListData struct {
|
||||
Uid int64 `json:"uid"`
|
||||
@@ -68,27 +53,36 @@ func (s *sGameKv) SavesV1() (err error) {
|
||||
var list []*ListData
|
||||
// 初始化列表,长度与keys数组一致
|
||||
list = make([]*ListData, 0)
|
||||
|
||||
// 从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
|
||||
}
|
||||
|
||||
//需要删除的key
|
||||
var delKey []string
|
||||
|
||||
// 遍历keys,获取每个用户的数据并填充到list中
|
||||
for _, cacheKey := range keys {
|
||||
//g.Log().Infof(ctx, "保存用户kv数据%v", v)
|
||||
//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 {
|
||||
@@ -98,40 +92,46 @@ 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,
|
||||
})
|
||||
|
||||
delKey = append(delKey, cacheKey)
|
||||
}
|
||||
|
||||
// 将列表数据保存到数据库
|
||||
if len(list) > 0 {
|
||||
_, err2 := g.Model("game_kv").Batch(30).Data(list).Save()
|
||||
list = make([]*ListData, 0)
|
||||
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 delKey {
|
||||
_, err2 = g.Redis().Del(ctx, v)
|
||||
if err2 != nil {
|
||||
g.Log().Errorf(ctx, "删除存档失败:%v,err=%v", v, err2)
|
||||
//删除当前key
|
||||
for _, v := range list {
|
||||
s.DelCacheKey(ctx, v.Uid)
|
||||
}
|
||||
list = make([]*ListData, 0)
|
||||
}
|
||||
|
||||
delKey = make([]string, 0)
|
||||
|
||||
}
|
||||
if err != nil {
|
||||
g.Log().Error(ctx, "当前kv数据入库失败: %v", err)
|
||||
}
|
||||
|
||||
return
|
||||
})
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// 删除缓存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 {
|
||||
g.Log().Error(ctx, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -60,7 +60,7 @@ type sSystemCron struct {
|
||||
func New() *sSystemCron {
|
||||
return &sSystemCron{
|
||||
taskChan: make(chan func(context.Context) error, 2),
|
||||
TaskTimeout: time.Minute * 30,
|
||||
TaskTimeout: time.Minute * 60,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -333,7 +333,7 @@ func (s *sSystemCron) RunFuncChan() {
|
||||
//ctx := gctx.New()
|
||||
func() {
|
||||
//超时释放资源
|
||||
ctx, cancel := context.WithTimeout(context.Background(), s.TaskTimeout)
|
||||
ctx, cancel := context.WithTimeout(gctx.New(), s.TaskTimeout)
|
||||
defer cancel()
|
||||
|
||||
// 使用匿名函数包裹来捕获 panic
|
||||
|
||||
@@ -31,7 +31,7 @@ type DataType struct {
|
||||
Url string `json:"url"` // S3 服务的访问 URL
|
||||
BucketName string `json:"bucket_name"` // 默认存储桶名称
|
||||
BucketNameCdn string `json:"bucket_name_cdn"` // CDN 存储桶名称
|
||||
Provider string `json:"provider"` // S3 服务的提供方
|
||||
//Provider string `json:"provider"` // S3 服务的提供方
|
||||
}
|
||||
|
||||
// Mod 定义了 S3 模块的结构体,包含一个 S3 客户端实例和配置信息
|
||||
@@ -184,6 +184,12 @@ func (s *Mod) ListObjects(bucketName string, prefix string) (res <-chan minio.Ob
|
||||
return
|
||||
}
|
||||
|
||||
// StatObject 获取指定存储桶中指定文件的元数据信息
|
||||
func (s *Mod) StatObject(bucketName string, objectName string) (res minio.ObjectInfo, err error) {
|
||||
res, err = s.client.StatObject(ctx, bucketName, objectName, minio.StatObjectOptions{})
|
||||
return
|
||||
}
|
||||
|
||||
// SetBucketPolicy 设置指定存储桶或对象前缀的访问策略
|
||||
// 目前使用固定的策略,可根据需求修改
|
||||
func (s *Mod) SetBucketPolicy(bucketName string, prefix string) (err error) {
|
||||
|
||||
@@ -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,48 @@ 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) (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缓存
|
||||
RefreshGetRedDotCache(uid int64)
|
||||
Del(uid int64, actId int)
|
||||
|
||||
@@ -5,6 +5,10 @@
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
type (
|
||||
IGameKv interface {
|
||||
// SavesV1 方法
|
||||
@@ -13,6 +17,8 @@ type (
|
||||
// @receiver s: sGameKv的实例。
|
||||
// @return err: 错误信息,如果操作成功,则为nil。
|
||||
SavesV1() (err error)
|
||||
// 删除缓存key
|
||||
DelCacheKey(ctx context.Context, uid int64)
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@@ -38,8 +38,8 @@ type (
|
||||
// @receiver s: sSystemCron的实例,代表一个调度系统。
|
||||
// @param typ: 任务的类型,决定该任务将被添加到哪个列表中。对应不同的时间间隔。
|
||||
// @param _func: 要添加的任务函数,该函数执行时应该返回一个error。
|
||||
// @param unique: 是否只在唯一服务器上执行
|
||||
AddCronV2(typ v1.CronType, _func func(context.Context) error, unique ...bool)
|
||||
// @param _onlyMain: 是否只在主服务器上执行一次,true 唯一执行,false 全局执行不判断唯一
|
||||
AddCronV2(typ v1.CronType, _func func(context.Context) error, _onlyMain ...bool)
|
||||
// StartCron 开始计划任务执行
|
||||
//
|
||||
// @Description:
|
||||
|
||||
@@ -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,9 +66,11 @@ 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 {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
//这个要放在最后
|
||||
|
||||
Reference in New Issue
Block a user