405 lines
11 KiB
Go
405 lines
11 KiB
Go
package cron
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/robfig/cron/v3"
|
|
"go.uber.org/zap"
|
|
|
|
"tyapi-server/internal/application/statistics"
|
|
)
|
|
|
|
// StatisticsCronJob 统计定时任务
|
|
type StatisticsCronJob struct {
|
|
appService statistics.StatisticsApplicationService
|
|
logger *zap.Logger
|
|
cron *cron.Cron
|
|
}
|
|
|
|
// NewStatisticsCronJob 创建统计定时任务
|
|
func NewStatisticsCronJob(
|
|
appService statistics.StatisticsApplicationService,
|
|
logger *zap.Logger,
|
|
) *StatisticsCronJob {
|
|
return &StatisticsCronJob{
|
|
appService: appService,
|
|
logger: logger,
|
|
cron: cron.New(cron.WithLocation(time.UTC)),
|
|
}
|
|
}
|
|
|
|
// Start 启动定时任务
|
|
func (j *StatisticsCronJob) Start() error {
|
|
j.logger.Info("启动统计定时任务")
|
|
|
|
// 每小时聚合任务 - 每小时的第5分钟执行
|
|
_, err := j.cron.AddFunc("5 * * * *", j.hourlyAggregationJob)
|
|
if err != nil {
|
|
return fmt.Errorf("添加小时聚合任务失败: %w", err)
|
|
}
|
|
|
|
// 每日聚合任务 - 每天凌晨1点执行
|
|
_, err = j.cron.AddFunc("0 1 * * *", j.dailyAggregationJob)
|
|
if err != nil {
|
|
return fmt.Errorf("添加日聚合任务失败: %w", err)
|
|
}
|
|
|
|
// 每周聚合任务 - 每周一凌晨2点执行
|
|
_, err = j.cron.AddFunc("0 2 * * 1", j.weeklyAggregationJob)
|
|
if err != nil {
|
|
return fmt.Errorf("添加周聚合任务失败: %w", err)
|
|
}
|
|
|
|
// 每月聚合任务 - 每月1号凌晨3点执行
|
|
_, err = j.cron.AddFunc("0 3 1 * *", j.monthlyAggregationJob)
|
|
if err != nil {
|
|
return fmt.Errorf("添加月聚合任务失败: %w", err)
|
|
}
|
|
|
|
// 数据清理任务 - 每天凌晨4点执行
|
|
_, err = j.cron.AddFunc("0 4 * * *", j.dataCleanupJob)
|
|
if err != nil {
|
|
return fmt.Errorf("添加数据清理任务失败: %w", err)
|
|
}
|
|
|
|
// 缓存预热任务 - 每天早上6点执行
|
|
_, err = j.cron.AddFunc("0 6 * * *", j.cacheWarmupJob)
|
|
if err != nil {
|
|
return fmt.Errorf("添加缓存预热任务失败: %w", err)
|
|
}
|
|
|
|
// 启动定时器
|
|
j.cron.Start()
|
|
|
|
j.logger.Info("统计定时任务启动成功")
|
|
return nil
|
|
}
|
|
|
|
// Stop 停止定时任务
|
|
func (j *StatisticsCronJob) Stop() {
|
|
j.logger.Info("停止统计定时任务")
|
|
j.cron.Stop()
|
|
j.logger.Info("统计定时任务已停止")
|
|
}
|
|
|
|
// ================ 定时任务实现 ================
|
|
|
|
// hourlyAggregationJob 小时聚合任务
|
|
func (j *StatisticsCronJob) hourlyAggregationJob() {
|
|
ctx := context.Background()
|
|
now := time.Now()
|
|
|
|
// 聚合上一小时的数据
|
|
lastHour := now.Add(-1 * time.Hour).Truncate(time.Hour)
|
|
|
|
j.logger.Info("开始执行小时聚合任务", zap.Time("target_hour", lastHour))
|
|
|
|
err := j.appService.ProcessHourlyAggregation(ctx, lastHour)
|
|
if err != nil {
|
|
j.logger.Error("小时聚合任务执行失败",
|
|
zap.Time("target_hour", lastHour),
|
|
zap.Error(err))
|
|
return
|
|
}
|
|
|
|
j.logger.Info("小时聚合任务执行成功", zap.Time("target_hour", lastHour))
|
|
}
|
|
|
|
// dailyAggregationJob 日聚合任务
|
|
func (j *StatisticsCronJob) dailyAggregationJob() {
|
|
ctx := context.Background()
|
|
now := time.Now()
|
|
|
|
// 聚合昨天的数据
|
|
yesterday := now.AddDate(0, 0, -1).Truncate(24 * time.Hour)
|
|
|
|
j.logger.Info("开始执行日聚合任务", zap.Time("target_date", yesterday))
|
|
|
|
err := j.appService.ProcessDailyAggregation(ctx, yesterday)
|
|
if err != nil {
|
|
j.logger.Error("日聚合任务执行失败",
|
|
zap.Time("target_date", yesterday),
|
|
zap.Error(err))
|
|
return
|
|
}
|
|
|
|
j.logger.Info("日聚合任务执行成功", zap.Time("target_date", yesterday))
|
|
}
|
|
|
|
// weeklyAggregationJob 周聚合任务
|
|
func (j *StatisticsCronJob) weeklyAggregationJob() {
|
|
ctx := context.Background()
|
|
now := time.Now()
|
|
|
|
// 聚合上一周的数据
|
|
lastWeek := now.AddDate(0, 0, -7).Truncate(24 * time.Hour)
|
|
|
|
j.logger.Info("开始执行周聚合任务", zap.Time("target_week", lastWeek))
|
|
|
|
err := j.appService.ProcessWeeklyAggregation(ctx, lastWeek)
|
|
if err != nil {
|
|
j.logger.Error("周聚合任务执行失败",
|
|
zap.Time("target_week", lastWeek),
|
|
zap.Error(err))
|
|
return
|
|
}
|
|
|
|
j.logger.Info("周聚合任务执行成功", zap.Time("target_week", lastWeek))
|
|
}
|
|
|
|
// monthlyAggregationJob 月聚合任务
|
|
func (j *StatisticsCronJob) monthlyAggregationJob() {
|
|
ctx := context.Background()
|
|
now := time.Now()
|
|
|
|
// 聚合上个月的数据
|
|
lastMonth := now.AddDate(0, -1, 0).Truncate(24 * time.Hour)
|
|
|
|
j.logger.Info("开始执行月聚合任务", zap.Time("target_month", lastMonth))
|
|
|
|
err := j.appService.ProcessMonthlyAggregation(ctx, lastMonth)
|
|
if err != nil {
|
|
j.logger.Error("月聚合任务执行失败",
|
|
zap.Time("target_month", lastMonth),
|
|
zap.Error(err))
|
|
return
|
|
}
|
|
|
|
j.logger.Info("月聚合任务执行成功", zap.Time("target_month", lastMonth))
|
|
}
|
|
|
|
// dataCleanupJob 数据清理任务
|
|
func (j *StatisticsCronJob) dataCleanupJob() {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("开始执行数据清理任务")
|
|
|
|
err := j.appService.CleanupExpiredData(ctx)
|
|
if err != nil {
|
|
j.logger.Error("数据清理任务执行失败", zap.Error(err))
|
|
return
|
|
}
|
|
|
|
j.logger.Info("数据清理任务执行成功")
|
|
}
|
|
|
|
// cacheWarmupJob 缓存预热任务
|
|
func (j *StatisticsCronJob) cacheWarmupJob() {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("开始执行缓存预热任务")
|
|
|
|
// 预热仪表板数据
|
|
err := j.warmupDashboardCache(ctx)
|
|
if err != nil {
|
|
j.logger.Error("仪表板缓存预热失败", zap.Error(err))
|
|
}
|
|
|
|
// 预热实时指标
|
|
err = j.warmupRealtimeMetricsCache(ctx)
|
|
if err != nil {
|
|
j.logger.Error("实时指标缓存预热失败", zap.Error(err))
|
|
}
|
|
|
|
j.logger.Info("缓存预热任务执行完成")
|
|
}
|
|
|
|
// ================ 缓存预热辅助方法 ================
|
|
|
|
// warmupDashboardCache 预热仪表板缓存
|
|
func (j *StatisticsCronJob) warmupDashboardCache(ctx context.Context) error {
|
|
// 获取所有用户角色
|
|
userRoles := []string{"admin", "user", "manager", "analyst"}
|
|
|
|
for _, role := range userRoles {
|
|
// 获取仪表板数据
|
|
query := &statistics.GetDashboardDataQuery{
|
|
UserRole: role,
|
|
Period: "today",
|
|
StartDate: time.Now().Truncate(24 * time.Hour),
|
|
EndDate: time.Now(),
|
|
}
|
|
|
|
_, err := j.appService.GetDashboardData(ctx, query)
|
|
if err != nil {
|
|
j.logger.Error("预热仪表板缓存失败",
|
|
zap.String("user_role", role),
|
|
zap.Error(err))
|
|
continue
|
|
}
|
|
|
|
j.logger.Info("仪表板缓存预热成功", zap.String("user_role", role))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// warmupRealtimeMetricsCache 预热实时指标缓存
|
|
func (j *StatisticsCronJob) warmupRealtimeMetricsCache(ctx context.Context) error {
|
|
// 获取所有指标类型
|
|
metricTypes := []string{"api_calls", "users", "finance", "products", "certification"}
|
|
|
|
for _, metricType := range metricTypes {
|
|
// 获取实时指标
|
|
query := &statistics.GetRealtimeMetricsQuery{
|
|
MetricType: metricType,
|
|
TimeRange: "last_hour",
|
|
}
|
|
|
|
_, err := j.appService.GetRealtimeMetrics(ctx, query)
|
|
if err != nil {
|
|
j.logger.Error("预热实时指标缓存失败",
|
|
zap.String("metric_type", metricType),
|
|
zap.Error(err))
|
|
continue
|
|
}
|
|
|
|
j.logger.Info("实时指标缓存预热成功", zap.String("metric_type", metricType))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ================ 手动触发任务 ================
|
|
|
|
// TriggerHourlyAggregation 手动触发小时聚合
|
|
func (j *StatisticsCronJob) TriggerHourlyAggregation(targetHour time.Time) error {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("手动触发小时聚合任务", zap.Time("target_hour", targetHour))
|
|
|
|
err := j.appService.ProcessHourlyAggregation(ctx, targetHour)
|
|
if err != nil {
|
|
j.logger.Error("手动小时聚合任务执行失败",
|
|
zap.Time("target_hour", targetHour),
|
|
zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
j.logger.Info("手动小时聚合任务执行成功", zap.Time("target_hour", targetHour))
|
|
return nil
|
|
}
|
|
|
|
// TriggerDailyAggregation 手动触发日聚合
|
|
func (j *StatisticsCronJob) TriggerDailyAggregation(targetDate time.Time) error {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("手动触发日聚合任务", zap.Time("target_date", targetDate))
|
|
|
|
err := j.appService.ProcessDailyAggregation(ctx, targetDate)
|
|
if err != nil {
|
|
j.logger.Error("手动日聚合任务执行失败",
|
|
zap.Time("target_date", targetDate),
|
|
zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
j.logger.Info("手动日聚合任务执行成功", zap.Time("target_date", targetDate))
|
|
return nil
|
|
}
|
|
|
|
// TriggerWeeklyAggregation 手动触发周聚合
|
|
func (j *StatisticsCronJob) TriggerWeeklyAggregation(targetWeek time.Time) error {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("手动触发周聚合任务", zap.Time("target_week", targetWeek))
|
|
|
|
err := j.appService.ProcessWeeklyAggregation(ctx, targetWeek)
|
|
if err != nil {
|
|
j.logger.Error("手动周聚合任务执行失败",
|
|
zap.Time("target_week", targetWeek),
|
|
zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
j.logger.Info("手动周聚合任务执行成功", zap.Time("target_week", targetWeek))
|
|
return nil
|
|
}
|
|
|
|
// TriggerMonthlyAggregation 手动触发月聚合
|
|
func (j *StatisticsCronJob) TriggerMonthlyAggregation(targetMonth time.Time) error {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("手动触发月聚合任务", zap.Time("target_month", targetMonth))
|
|
|
|
err := j.appService.ProcessMonthlyAggregation(ctx, targetMonth)
|
|
if err != nil {
|
|
j.logger.Error("手动月聚合任务执行失败",
|
|
zap.Time("target_month", targetMonth),
|
|
zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
j.logger.Info("手动月聚合任务执行成功", zap.Time("target_month", targetMonth))
|
|
return nil
|
|
}
|
|
|
|
// TriggerDataCleanup 手动触发数据清理
|
|
func (j *StatisticsCronJob) TriggerDataCleanup() error {
|
|
ctx := context.Background()
|
|
|
|
j.logger.Info("手动触发数据清理任务")
|
|
|
|
err := j.appService.CleanupExpiredData(ctx)
|
|
if err != nil {
|
|
j.logger.Error("手动数据清理任务执行失败", zap.Error(err))
|
|
return err
|
|
}
|
|
|
|
j.logger.Info("手动数据清理任务执行成功")
|
|
return nil
|
|
}
|
|
|
|
// TriggerCacheWarmup 手动触发缓存预热
|
|
func (j *StatisticsCronJob) TriggerCacheWarmup() error {
|
|
j.logger.Info("手动触发缓存预热任务")
|
|
|
|
// 预热仪表板缓存
|
|
err := j.warmupDashboardCache(context.Background())
|
|
if err != nil {
|
|
j.logger.Error("手动仪表板缓存预热失败", zap.Error(err))
|
|
}
|
|
|
|
// 预热实时指标缓存
|
|
err = j.warmupRealtimeMetricsCache(context.Background())
|
|
if err != nil {
|
|
j.logger.Error("手动实时指标缓存预热失败", zap.Error(err))
|
|
}
|
|
|
|
j.logger.Info("手动缓存预热任务执行完成")
|
|
return nil
|
|
}
|
|
|
|
// ================ 任务状态查询 ================
|
|
|
|
// GetCronEntries 获取定时任务条目
|
|
func (j *StatisticsCronJob) GetCronEntries() []cron.Entry {
|
|
return j.cron.Entries()
|
|
}
|
|
|
|
// GetNextRunTime 获取下次运行时间
|
|
func (j *StatisticsCronJob) GetNextRunTime() time.Time {
|
|
entries := j.cron.Entries()
|
|
if len(entries) == 0 {
|
|
return time.Time{}
|
|
}
|
|
|
|
// 返回最近的运行时间
|
|
nextRun := entries[0].Next
|
|
for _, entry := range entries[1:] {
|
|
if entry.Next.Before(nextRun) {
|
|
nextRun = entry.Next
|
|
}
|
|
}
|
|
|
|
return nextRun
|
|
}
|
|
|
|
// IsRunning 检查任务是否正在运行
|
|
func (j *StatisticsCronJob) IsRunning() bool {
|
|
return j.cron != nil
|
|
}
|
|
|