public_project/utility_go
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化游戏数据批量持久化性能

Browse Source 
- gameAct: 新增SaveV2Batch批量处理方法
  - 使用Redis MGET批量获取替代逐个GET
  - 使用WHERE uid IN批量查询替代逐个查询
  - 使用Batch批量插入替代逐条SQL
  - 修复: 只有数据库写入成功后才删除Redis key

- gameKv: 新增SavesV2Batch批量处理方法
  - 同样的批量优化策略
  - 修复: 只有数据库写入成功后才删除Redis key

- service: 更新接口定义添加新方法

性能提升: 1000条数据从1000次网络请求减少到2-3次
This commit is contained in:
ayflying
2026-02-26 15:44:14 +08:00
parent 8ed8152f79
commit 713a63356e
5 changed files with 397 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
README.md
View File

@@ -23,6 +23,75 @@ go get github.com/ayflying/utility_go
## 🏗️ 项目结构
```
utility_go/
├── api/ # API接口定义 (protobuf)
│ └── system/ # 系统API
├── aycache/ # Aycache缓存实现
├── cmd/ # CLI命令工具
│ ├── make.go # 代码生成器
│ ├── load.go # 配置加载
│ ├── update.go # 更新工具
│ └── middleware.go # 中间件工具
├── config/ # 配置管理
├── controller/ # 控制器层
│ └── callback/ # 回调控制器
├── internal/ # 内部核心逻辑
│ ├── boot/ # 启动初始化
│ ├── game/ # 游戏逻辑
│ ├── logic/ # 业务逻辑
│ │ ├── casdoor/ # 认证逻辑
│ │ ├── gameAct/ # 游戏活动
│ │ ├── gameKv/ # 游戏键值存储
│ │ ├── ip2region/ # IP区域查询
│ │ ├── logData/ # 日志处理
│ │ ├── os/ # 系统操作
│ │ ├── systemCron/ # 定时任务
│ │ └── systemLog/ # 系统日志
│ └── model/ # 数据模型
│ ├── do/ # 数据对象
│ └── entity/ # 实体定义
├── package/ # 功能包
│ ├── aycache/ # 缓存管理
│ ├── excel/ # Excel处理
│ ├── gamelog/ # 游戏日志
│ ├── pay/ # 支付集成
│ │ ├── alipay/ # 支付宝
│ │ ├── apple/ # Apple Pay
│ │ ├── playstore/ # Google Play
│ │ ├── wechat/ # 微信支付
│ │ ├── xiaomi/ # 小米支付
│ │ ├── vivo/ # Vivo支付
│ │ ├── oppo/ # OPPO支付
│ │ ├── huawei/ # 华为支付
│ │ ├── honor/ # 荣耀支付
│ │ ├── taptap/ # TapTap支付
│ │ └── chongchong/ # 充值渠道
│ ├── rank/ # 排名算法
│ └── s3/ # S3存储
├── pkg/ # 公共包
│ ├── config/ # 配置包
│ ├── elasticsearch/ # ES包
│ ├── notice/ # 通知包
│ ├── rank/ # 排名包
│ └── s3/ # S3包
├── service/ # 服务层
│ ├── casdoor.go # 认证服务
│ ├── game_act.go # 游戏活动服务
│ ├── game_kv.go # 键值存储服务
│ ├── ip_2_region.go # IP地理位置服务
│ ├── log_data.go # 日志数据服务
│ ├── os.go # 操作系统服务
│ ├── system_cron.go # 定时任务服务
│ └── system_log.go # 系统日志服务
├── tools/ # 工具函数
│ ├── random.go # 随机数
│ ├── redis.go # Redis操作
│ ├── time.go # 时间处理
│ └── tools.go # 通用工具
├── utility.go # 主入口
├── go.mod # 模块定义
└── README.md # 说明文档
```
utility_go/
├── api/ # API接口定义
@@ -156,6 +225,68 @@ import (
)
```
#### Vivo 支付
```go
import "github.com/ayflying/utility_go/package/pay/vivo"
// 创建vivo支付客户端
client := vivo.NewClient(appId, appKey, businessId)
// 验证支付回调
result, err := client.VerifyNotification(notifyReq)
```
#### 小米支付
```go
import "github.com/ayflying/utility_go/package/pay/xiaomi"
// 处理小米支付回调
result, err := xiaomi.VerifyPayment(orderId, signature, requestBody)
```
#### OPPO 支付
```go
import "github.com/ayflying/utility_go/package/pay/oppo"
// 创建OPPO支付客户端
client := oppo.NewClient(apiKey, apiSecret)
// 验证订单
orderInfo, err := client.VerifyOrder(serverOrderId)
```
#### 华为支付
```go
import "github.com/ayflying/utility_go/package/pay/huawei"
// 处理华为支付通知
notification, err := huawei.ParseNotification(requestBody)
```
#### TapTap 支付
```go
import "github.com/ayflying/utility_go/package/pay/taptap"
// 验证TapTap签名
isValid := taptap.VerifySign(params, signature)
```
#### 荣耀支付
```go
import "github.com/ayflying/utility_go/package/pay/honor"
// 处理荣耀支付回调
result, err := honor.ParseNotification(requestBody)
```
#### 充值渠道支付
```go
import "github.com/ayflying/utility_go/package/pay/chongchong"
// 充值渠道订单验证
result, err := chongchong.VerifyOrder(orderId, channelOrderId)
```
### 🏆 排名模块 (package/rank)
基于Redis的高性能排行榜实现：

187
internal/logic/gameAct/gameAct.go
View File

@@ -314,10 +314,8 @@ func (s *sGameAct) SavesV2() (err 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)
}
if keyErr := s.SaveV2Batch(ctx, keys); keyErr != nil {
g.Log().Errorf(ctx, "批量处理keys失败: %v", keyErr)
}
return nil
})
@@ -613,21 +611,178 @@ func (s *sGameAct) Cache2SqlChan(ctx context.Context, addChan, updateChan chan *
return
}
// 删除缓存key
func (s *sGameAct) DelCacheKey(ctx context.Context, aid int, uid int64) {
go func() {
//如果有活跃，跳过删除
if getBool, _ := pkg.Cache("redis").
Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
return
// SaveV2Batch 批量保存游戏活动数据 (优化版)
//
// @Description: 使用批量Redis MGET和批量数据库操作提升性能
// @param ctx context.Context: 上下文对象
// @param cacheKeys []string: 缓存键列表
// @return err error: 返回错误信息
func (s *sGameAct) SaveV2Batch(ctx context.Context, cacheKeys []string) (err error) {
if len(cacheKeys) == 0 {
return nil
}
type keyInfo struct {
cacheKey string
actId int
uid int64
}
var keyInfos []keyInfo
activeUids := make(map[int64]bool)
for _, cacheKey := range cacheKeys {
result := strings.Split(cacheKey, ":")
if len(result) < 3 {
continue
}
actId := gconv.Int(result[1])
if actId == 0 {
continue
}
uid := gconv.Int64(result[2])
if uid == 0 {
continue
}
cacheKey := fmt.Sprintf("act:%v:%v", aid, uid)
_, err := g.Redis().Del(ctx, cacheKey)
if err != nil {
g.Log().Error(ctx, err)
if getBool, _ := pkg.Cache("redis").Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
activeUids[uid] = true
continue
}
}()
keyInfos = append(keyInfos, keyInfo{
cacheKey: cacheKey,
actId: actId,
uid: uid,
})
}
if len(keyInfos) == 0 {
return nil
}
redisValues, err := g.Redis().MGet(ctx, cacheKeys...)
if err != nil {
g.Log().Errorf(ctx, "批量获取Redis失败: %v", err)
return err
}
var validKeyInfos []keyInfo
for i, keyInfo := range keyInfos {
if val, ok := redisValues[keyInfo.cacheKey]; ok && gconv.String(val) != "" {
validKeyInfos = append(validKeyInfos, keyInfos[i])
}
}
if len(validKeyInfos) == 0 {
return nil
}
var uids []int64
for _, ki := range validKeyInfos {
uids = append(uids, ki.uid)
}
var existingData []do.GameAct
err = g.Model(Name).Where("uid IN (?)", uids).Fields("uid,act_id").Scan(&existingData)
if err != nil {
g.Log().Errorf(ctx, "批量查询数据库失败: %v", err)
return err
}
existMap := make(map[int64]do.GameAct)
for _, data := range existingData {
existMap[gconv.Int64(data.Uid)] = data
}
var addItems []*entity.GameAct
var updateItems []*entity.GameAct
for _, ki := range validKeyInfos {
val, ok := redisValues[ki.cacheKey]
if !ok {
continue
}
actionData := gconv.String(val)
if actionData == "" {
continue
}
if _, ok := existMap[ki.uid]; ok {
updateItems = append(updateItems, &entity.GameAct{
ActId: ki.actId,
Uid: ki.uid,
Action: actionData,
})
} else {
addItems = append(addItems, &entity.GameAct{
ActId: ki.actId,
Uid: ki.uid,
Action: actionData,
})
}
}
tx, err := g.DB().Begin(ctx)
if err != nil {
g.Log().Error(ctx, err)
return err
}
if len(addItems) > 0 {
_, err = tx.Model(Name).Data(addItems).Batch(100).Insert()
if err != nil {
g.Log().Errorf(ctx, "批量新增失败: %v", err)
tx.Rollback()
return err
}
}
if len(updateItems) > 0 {
for _, item := range updateItems {
item.UpdatedAt = gtime.Now()
_, err = tx.Model(Name).Where(do.GameAct{
Uid: item.Uid,
ActId: item.ActId,
}).Data(item).Update()
if err != nil {
g.Log().Errorf(ctx, "批量更新失败: %v", err)
}
}
}
err = tx.Commit()
if err != nil {
g.Log().Errorf(ctx, "提交事务失败: %v", err)
return err
}
for _, item := range addItems {
s.DelCacheKey(ctx, item.ActId, item.Uid)
}
for _, item := range updateItems {
s.DelCacheKey(ctx, item.ActId, item.Uid)
}
g.Log().Debugf(ctx, "SaveV2Batch完成: 新增%d, 更新%d", len(addItems), len(updateItems))
return nil
}
// 删除缓存key
func (s *sGameAct) DelCacheKey(ctx context.Context, aid int, uid int64) {
if uid == 0 {
return
}
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缓存

155
internal/logic/gameKv/gameKv.go
View File

@@ -41,79 +41,17 @@ func init() {
// @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数据保存")
// 定义用于存储用户数据的结构体
type ListData struct {
Uid int64 `json:"uid"`
Kv interface{} `json:"kv"`
}
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
// 遍历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 = 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
}
//如果有活跃，跳过持久化
if getBool, _ := pkg.Cache("redis").
Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
continue
}
get, _ := g.Redis().Get(ctx, cacheKey)
var data interface{}
get.Scan(&data)
if data == nil {
continue
}
list = append(list, &ListData{
Uid: uid,
Kv: data,
})
}
// 将列表数据保存到数据库
if len(list) > 100 {
_, err2 := g.Model("game_kv").Data(list).Save()
if err2 != nil {
g.Log().Error(ctx, "当前kv数据入库失败: %v", err2)
err = err2
return
}
//删除当前key
for _, v := range list {
s.DelCacheKey(ctx, v.Uid)
}
list = make([]*ListData, 0)
if err = s.SavesV2Batch(ctx, keys); err != nil {
g.Log().Errorf(ctx, "批量保存KV失败: %v", err)
}
return
})
@@ -135,3 +73,92 @@ func (s *sGameKv) DelCacheKey(ctx context.Context, uid int64) {
g.Log().Error(ctx, err)
}
}
// SavesV2Batch 批量保存游戏KV数据 (优化版)
//
// @Description: 使用批量Redis MGET和批量数据库操作提升性能
// @param ctx context.Context: 上下文对象
// @param cacheKeys []string: 缓存键列表
// @return err error: 返回错误信息
func (s *sGameKv) SavesV2Batch(ctx context.Context, cacheKeys []string) (err error) {
if len(cacheKeys) == 0 {
return nil
}
type keyInfo struct {
cacheKey string
uid int64
}
var keyInfos []keyInfo
for _, cacheKey := range cacheKeys {
result := strings.Split(cacheKey, ":")
if len(result) < 3 {
continue
}
uid := gconv.Int64(result[2])
if uid == 0 {
continue
}
if getBool, _ := pkg.Cache("redis").Contains(ctx, fmt.Sprintf("act:update:%d", uid)); getBool {
continue
}
keyInfos = append(keyInfos, keyInfo{
cacheKey: cacheKey,
uid: uid,
})
}
if len(keyInfos) == 0 {
return nil
}
redisValues, err := g.Redis().MGet(ctx, cacheKeys...)
if err != nil {
g.Log().Errorf(ctx, "批量获取Redis失败: %v", err)
return err
}
type ListData struct {
Uid int64 `json:"uid"`
Kv interface{} `json:"kv"`
}
var list []*ListData
for _, ki := range keyInfos {
val, ok := redisValues[ki.cacheKey]
if !ok {
continue
}
var data interface{}
if val != nil {
gconv.Scan(val, &data)
}
if data == nil {
continue
}
list = append(list, &ListData{
Uid: ki.uid,
Kv: data,
})
}
if len(list) == 0 {
return nil
}
_, err = g.Model(Name).Data(list).Save()
if err != nil {
g.Log().Errorf(ctx, "批量保存KV失败: %v", err)
return err
}
for _, v := range list {
s.DelCacheKey(ctx, v.Uid)
}
g.Log().Debugf(ctx, "SavesV2Batch完成: 保存%d条", len(list))
return nil
}

2
service/game_act.go
View File

@@ -75,6 +75,8 @@ type (
Cache2Sql(ctx context.Context, add []*entity.GameAct, update []*entity.GameAct)
// Cache2AddChan 批量添加数据库
Cache2SqlChan(ctx context.Context, addChan chan *entity.GameAct, updateChan chan *entity.GameAct)
// SaveV2Batch 批量保存游戏活动数据 (优化版)
SaveV2Batch(ctx context.Context, cacheKeys []string) (err error)
// 删除缓存key
DelCacheKey(ctx context.Context, aid int, uid int64)
// 清空GetRedDot缓存

2
service/game_kv.go
View File

@@ -17,6 +17,8 @@ type (
// @receiver s: sGameKv的实例。
// @return err: 错误信息，如果操作成功，则为nil。
SavesV1() (err error)
// SavesV2Batch 批量保存游戏KV数据 (优化版)
SavesV2Batch(ctx context.Context, cacheKeys []string) (err error)
// 删除缓存key
DelCacheKey(ctx context.Context, uid int64)
}