act持久化事务最大提交从5改为100

internal/logic/gameAct/gameAct.go

@@ -26,7 +26,7 @@ var (
 	Name       = "game_act"
 	ActList    = gset.New(true)
 	RunTimeMax *gtime.Time
-	TaskMax    int64 = 5
+	TaskMax    int64 = 100
 )
 
 type sGameAct struct {
@@ -237,7 +237,7 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
 				}
 
 				//删除缓存
-				go s.DelCacheKey(ctx, v.ActId, v.Uid)
+				s.DelCacheKey(ctx, v.ActId, v.Uid)
 
 				updateCount++
 				update = make([]*entity.GameAct, 0)
@@ -265,10 +265,10 @@ func (s *sGameAct) Save(ctx context.Context, actId int) (err error) {
 				return
 			}
 
-			for _, v2 := range add {
-				//删除缓存
-				go s.DelCacheKey(ctx, v2.ActId, v2.Uid)
-			}
+			//for _, v2 := range add {
+			//	//删除缓存
+			//	s.DelCacheKey(ctx, v2.ActId, v2.Uid)
+			//}
 
 			//g.Log().Debugf(ctx, "当前 %v 写入数据库: %v 条", actId, count)
 			add = make([]*entity.GameAct, 0)
@@ -461,7 +461,7 @@ func (s *sGameAct) Cache2Sql(ctx context.Context, add, update []*entity.GameAct)
 			s.DelCacheKey(ctx, v.ActId, v.Uid)
 		}
 		//循环结束了，最后写入一波
-		g.Log().Debugf(ctx, "act当前更新数据库: %v 条", updateCount)
+		g.Log().Debugf(ctx, "Cache2Sql运行结束，act当前更新数据库: %v 条", updateCount)
 		update = (update)[:0]
 	}
 
@@ -479,7 +479,7 @@ func (s *sGameAct) Cache2Sql(ctx context.Context, add, update []*entity.GameAct)
 			}
 			addCount++
 			if addCount > TaskMax {
-				g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
+				g.Log().Debugf(ctx, "超过%v条，act当前写入数据库: %v 条", TaskMax, addCount)
 				err = tx.Commit()
 				if err != nil {
 					g.Log().Debugf(ctx, "act当前写入数据库失败:%v", err)
@@ -493,7 +493,7 @@ func (s *sGameAct) Cache2Sql(ctx context.Context, add, update []*entity.GameAct)
 		}
 
 		//循环结束了，最后写入一波
-		g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
+		g.Log().Debugf(ctx, "Cache2Sql运行结束，act当前写入数据库: %v 条", addCount)
 		add = (add)[:0]
 	}
 
@@ -510,6 +510,8 @@ func (s *sGameAct) Cache2SqlChan(ctx context.Context, addChan, updateChan chan *
 	//通道关闭标志
 	addClosed := false
 	updateClosed := false
+	//写入总数
+	var addAllCount, updateAllCount int64
 
 	tx, err := g.DB().Begin(ctx)
 	if err != nil {
@@ -545,12 +547,14 @@ func (s *sGameAct) Cache2SqlChan(ctx context.Context, addChan, updateChan chan *
 			addCount += row
 
 			if addCount > TaskMax {
-				g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
+				// g.Log().Debugf(ctx, "超过%v条，act当前写入数据库: %v 条", TaskMax, addCount)
 				err = tx.Commit()
 				if err != nil {
 					g.Log().Debugf(ctx, "act当前写入数据库失败:%v", err)
 					return
 				}
+				//清空数量前累加一下
+				addAllCount += addCount
 				addCount = 0
 				tx, err = g.DB().Begin(ctx)
 			}
@@ -579,12 +583,14 @@ func (s *sGameAct) Cache2SqlChan(ctx context.Context, addChan, updateChan chan *
 			updateCount++
 
 			if updateCount > TaskMax {
-				g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
+				// g.Log().Debugf(ctx, "超过%v条，act当前更新数据库: %v 条", TaskMax, updateCount)
 				err = tx.Commit()
 				if err != nil {
-					g.Log().Debugf(ctx, "act当前写入数据库失败:%v", err)
+					g.Log().Debugf(ctx, "act当前更新数据库失败:%v", err)
 					return
 				}
+				//清空数量前累加一下
+				updateAllCount += updateCount
 				updateCount = 0
 				tx, err = g.DB().Begin(ctx)
 			}
@@ -599,25 +605,29 @@ func (s *sGameAct) Cache2SqlChan(ctx context.Context, addChan, updateChan chan *
 	}
 
 	err = tx.Commit()
+	addAllCount += addCount
+	updateAllCount += updateCount
 	// 仅在所有通道处理完毕后打印最终计数（移除中间冗余日志）
-	g.Log().Debugf(ctx, "act当前写入数据库: %v 条", addCount)
-	g.Log().Debugf(ctx, "act当前更新数据库: %v 条", updateCount)
+	g.Log().Debugf(ctx, "运行结束act当前写入数据库: %v 条", addAllCount)
+	g.Log().Debugf(ctx, "运行结束act当前更新数据库: %v 条", updateAllCount)
 	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
-	}
+	go func() {
+		//如果有活跃，跳过删除
+		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)
-	}
+		cacheKey := fmt.Sprintf("act:%v:%v", aid, uid)
+		_, err := g.Redis().Del(ctx, cacheKey)
+		if err != nil {
+			g.Log().Error(ctx, err)
+		}
+	}()
 }
 
 // 清空GetRedDot缓存

package/gamelog/test/gamelog_test.go

@@ -16,7 +16,8 @@ func TestGamelog(t *testing.T) {
 		// 必填
 		Pid: "test5", // 项目ID
 		// BaseUrl: "http://47.76.178.47:10101", // 香港测试服上报地址
-		BaseUrl: "http://101.37.28.111:10101", // 香港测试服上报地址
+		// BaseUrl: "http://101.37.28.111:10101", // 香港测试服上报地址
+		BaseUrl: "http://47.77.200.131:10101", // 美国BIDB服务器
 		// BaseUrl:       "http://127.0.0.1:10101", // 本次测试上报地址
 		ReportSk:      "sngame2025", // xor混淆key
 		FlushInterval: 5,            // 上报间隔
@@ -52,3 +53,53 @@ func TestGamelog(t *testing.T) {
 		glsdk.Shutdown()
 	})
 }
+
+func TestPressMQ(t *testing.T) {
+	glsdk, err := gamelog.INIT(&gamelog.SDKConfig{
+		// 必填
+		Pid:           "yotest",                     // 项目ID
+		BaseUrl:       "http://47.77.200.131:10101", // 美国BIDB服务器
+		ReportSk:      "sngame2025",                 // xor混淆key
+		FlushInterval: 6,                            // 上报间隔
+		DiskBakPath:   "gamelog",                    // 本地磁盘备份, 用于意外情况下临时保存日志, 请确保该目录持久化(容器内要挂载). 每次启动时或每N次上报时加载到失败队列
+		// 可填
+		RetryN:       2,   // 默认每10次, 上传检查一次磁盘的失败数据
+		ChanSize:     500, // 默认1000, 信道size
+		SendSaveType: 2,   // 发送存储类型, 默认不设置为0代表文件存储, 2代表走kafka可实同步日志
+	})
+
+	// 随机测试事件和属性
+	events := []string{"e1", "e2", "e3", "e4"}
+	pms := []map[string]any{
+		{"a": "1"},
+		{"a": "2"},
+		{"a": "3"},
+		{"a": "4"},
+	}
+	uuids := []string{}
+	for i := 0; i < 100; i++ {
+		uuidval, _ := uuid.NewUUID()
+		randUid := strings.ReplaceAll(uuidval.String(), "-", "")
+		uuids = append(uuids, randUid)
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	n := 0
+	const limit = 30000
+	gtest.C(t, func(t *gtest.T) {
+		go func() {
+			for {
+				glsdk.LogLtz(uuids[grand.Intn(len(uuids))], events[grand.Intn(len(events))], pms[grand.Intn(len(pms))])
+				// 并发控制
+				n++
+				if n%limit == 0 {
+					time.Sleep(time.Second * 1)
+				}
+			}
+		}()
+		time.Sleep(time.Second * 120)
+		// 模拟等待信号后优雅关闭
+		glsdk.Shutdown()
+	})
+}