package polling import ( "context" "time" "github.com/redis/go-redis/v9" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/polling" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" ) // MonitoringService 轮询监控服务 type MonitoringService struct { redis *redis.Client queueMgr *polling.PollingQueueManager } // NewMonitoringService 创建轮询监控服务实例 // queueMgr 可选(nil 时降级到旧非分片 Key 兼容模式) func NewMonitoringService(redis *redis.Client, opts ...interface{}) *MonitoringService { svc := &MonitoringService{redis: redis} for _, opt := range opts { if qm, ok := opt.(*polling.PollingQueueManager); ok { svc.queueMgr = qm } } return svc } // NewMonitoringServiceWithQueueMgr 创建注入了 PollingQueueManager 的监控服务(推荐) func NewMonitoringServiceWithQueueMgr(redis *redis.Client, queueMgr *polling.PollingQueueManager, _ *zap.Logger) *MonitoringService { return &MonitoringService{redis: redis, queueMgr: queueMgr} } // OverviewStats 总览统计 type OverviewStats struct { TotalCards int64 `json:"total_cards"` // 总卡数 InitializedCards int64 `json:"initialized_cards"` // 已初始化卡数 InitProgress float64 `json:"init_progress"` // 初始化进度 (0-100) IsInitializing bool `json:"is_initializing"` // 是否正在初始化 RealnameQueueSize int64 `json:"realname_queue_size"` // 实名检查队列大小 CarddataQueueSize int64 `json:"carddata_queue_size"` // 流量检查队列大小 PackageQueueSize int64 `json:"package_queue_size"` // 套餐检查队列大小 } // QueueStatus 队列状态 type QueueStatus struct { TaskType string `json:"task_type"` // 任务类型 TaskTypeName string `json:"task_type_name"` // 任务类型名称 QueueSize int64 `json:"queue_size"` // 队列大小 ManualPending int64 `json:"manual_pending"` // 手动触发待处理数 DueCount int64 `json:"due_count"` // 到期待处理数 AvgWaitTime float64 `json:"avg_wait_time_s"` // 平均等待时间(秒) } // TaskStats 任务统计 type TaskStats struct { TaskType string `json:"task_type"` // 任务类型 TaskTypeName string `json:"task_type_name"` // 任务类型名称 SuccessCount1h int64 `json:"success_count_1h"` // 1小时成功数 FailureCount1h int64 `json:"failure_count_1h"` // 1小时失败数 TotalCount1h int64 `json:"total_count_1h"` // 1小时总数 SuccessRate float64 `json:"success_rate"` // 成功率 (0-100) AvgDurationMs float64 `json:"avg_duration_ms"` // 平均耗时(毫秒) } // InitProgress 初始化进度 type InitProgress struct { TotalCards int64 `json:"total_cards"` // 总卡数 InitializedCards int64 `json:"initialized_cards"` // 已初始化卡数 Progress float64 `json:"progress"` // 进度百分比 (0-100) IsComplete bool `json:"is_complete"` // 是否完成 StartedAt time.Time `json:"started_at"` // 开始时间 EstimatedETA string `json:"estimated_eta"` // 预计完成时间 } // GetOverview 获取总览统计 func (s *MonitoringService) GetOverview(ctx context.Context) (*OverviewStats, error) { stats := &OverviewStats{} // 获取初始化进度 progressKey := constants.RedisPollingInitProgressKey() progressData, err := s.redis.HGetAll(ctx, progressKey).Result() if err != nil && err != redis.Nil { return nil, errors.Wrap(errors.CodeRedisError, err, "获取初始化进度失败") } if total, ok := progressData["total"]; ok { stats.TotalCards = parseInt64(total) } if initialized, ok := progressData["initialized"]; ok { stats.InitializedCards = parseInt64(initialized) } if stats.TotalCards > 0 { stats.InitProgress = float64(stats.InitializedCards) / float64(stats.TotalCards) * 100 } stats.IsInitializing = stats.InitializedCards < stats.TotalCards && stats.TotalCards > 0 if s.queueMgr != nil { stats.RealnameQueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, constants.TaskTypePollingRealname) stats.CarddataQueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, constants.TaskTypePollingCarddata) stats.PackageQueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, constants.TaskTypePollingPackage) } else { stats.RealnameQueueSize, _ = s.redis.ZCard(ctx, constants.RedisPollingQueueRealnameKey()).Result() stats.CarddataQueueSize, _ = s.redis.ZCard(ctx, constants.RedisPollingQueueCarddataKey()).Result() stats.PackageQueueSize, _ = s.redis.ZCard(ctx, constants.RedisPollingQueuePackageKey()).Result() } return stats, nil } // GetQueueStatuses 获取所有队列状态 func (s *MonitoringService) GetQueueStatuses(ctx context.Context) ([]*QueueStatus, error) { taskTypes := pollingMonitorTaskTypes() result := make([]*QueueStatus, 0, len(taskTypes)) now := time.Now() for _, taskType := range taskTypes { status := &QueueStatus{ TaskType: taskType, TaskTypeName: s.getTaskTypeName(taskType), } // 获取队列 key var queueKey string switch taskType { case constants.TaskTypePollingRealname: queueKey = constants.RedisPollingQueueRealnameKey() case constants.TaskTypePollingCarddata: queueKey = constants.RedisPollingQueueCarddataKey() case constants.TaskTypePollingPackage: queueKey = constants.RedisPollingQueuePackageKey() } if s.queueMgr != nil { status.QueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, taskType) status.DueCount, _ = s.queueMgr.GetTotalDueCount(ctx, taskType, now) status.AvgWaitTime, _ = s.queueMgr.GetAverageWaitTime(ctx, taskType, now, 10) } else { status.QueueSize, _ = s.redis.ZCard(ctx, queueKey).Result() status.DueCount, _ = s.redis.ZCount(ctx, queueKey, "-inf", formatInt64(now.Unix())).Result() status.AvgWaitTime = s.getLegacyAverageWaitTime(ctx, queueKey, now.Unix()) } // 获取手动触发队列待处理数 manualKey := constants.RedisPollingManualQueueKey(taskType) status.ManualPending, _ = s.redis.LLen(ctx, manualKey).Result() result = append(result, status) } return result, nil } // getLegacyAverageWaitTime 从旧版非分片队列计算平均等待时间。 func (s *MonitoringService) getLegacyAverageWaitTime(ctx context.Context, queueKey string, now int64) float64 { earliest, err := s.redis.ZRangeWithScores(ctx, queueKey, 0, 9).Result() if err != nil || len(earliest) == 0 { return 0 } var totalWait float64 for _, z := range earliest { waitTime := float64(now) - z.Score if waitTime > 0 { totalWait += waitTime } } return totalWait / float64(len(earliest)) } // GetTaskStatuses 获取所有任务统计 func (s *MonitoringService) GetTaskStatuses(ctx context.Context) ([]*TaskStats, error) { taskTypes := pollingMonitorTaskTypes() result := make([]*TaskStats, 0, len(taskTypes)) for _, taskType := range taskTypes { stats := &TaskStats{ TaskType: taskType, TaskTypeName: s.getTaskTypeName(taskType), } // 获取统计数据 statsKey := constants.RedisPollingStatsKey(taskType) data, err := s.redis.HGetAll(ctx, statsKey).Result() if err != nil && err != redis.Nil { continue } if success, ok := data["success_count_1h"]; ok { stats.SuccessCount1h = parseInt64(success) } if failure, ok := data["failure_count_1h"]; ok { stats.FailureCount1h = parseInt64(failure) } stats.TotalCount1h = stats.SuccessCount1h + stats.FailureCount1h if stats.TotalCount1h > 0 { stats.SuccessRate = float64(stats.SuccessCount1h) / float64(stats.TotalCount1h) * 100 if duration, ok := data["total_duration_1h"]; ok { totalDuration := parseInt64(duration) stats.AvgDurationMs = float64(totalDuration) / float64(stats.TotalCount1h) } } result = append(result, stats) } return result, nil } // GetInitProgress 获取初始化进度详情 func (s *MonitoringService) GetInitProgress(ctx context.Context) (*InitProgress, error) { progressKey := constants.RedisPollingInitProgressKey() data, err := s.redis.HGetAll(ctx, progressKey).Result() if err != nil && err != redis.Nil { return nil, errors.Wrap(errors.CodeRedisError, err, "获取初始化进度失败") } progress := &InitProgress{} if total, ok := data["total"]; ok { progress.TotalCards = parseInt64(total) } if initialized, ok := data["initialized"]; ok { progress.InitializedCards = parseInt64(initialized) } if startedAt, ok := data["started_at"]; ok { if t, err := time.Parse(time.RFC3339, startedAt); err == nil { progress.StartedAt = t } } if progress.TotalCards > 0 { progress.Progress = float64(progress.InitializedCards) / float64(progress.TotalCards) * 100 } progress.IsComplete = progress.InitializedCards >= progress.TotalCards && progress.TotalCards > 0 // 估算完成时间 if !progress.IsComplete && progress.InitializedCards > 0 && !progress.StartedAt.IsZero() { elapsed := time.Since(progress.StartedAt) remaining := progress.TotalCards - progress.InitializedCards rate := float64(progress.InitializedCards) / elapsed.Seconds() if rate > 0 { etaSeconds := float64(remaining) / rate eta := time.Now().Add(time.Duration(etaSeconds) * time.Second) progress.EstimatedETA = eta.Format("15:04:05") } } return progress, nil } // getTaskTypeName 获取任务类型的中文名称 func (s *MonitoringService) getTaskTypeName(taskType string) string { switch taskType { case constants.TaskTypePollingRealname: return "实名检查" case constants.TaskTypePollingCarddata: return "流量检查" case constants.TaskTypePollingPackage: return "套餐检查" case constants.TaskTypePollingProtect: return "保护期检查" case constants.TaskTypePollingCardStatus: return "卡状态检查" default: return taskType } } // pollingMonitorTaskTypes 返回后台监控需要展示的全部轮询任务类型。 func pollingMonitorTaskTypes() []string { return []string{ constants.TaskTypePollingRealname, constants.TaskTypePollingCarddata, constants.TaskTypePollingPackage, constants.TaskTypePollingProtect, constants.TaskTypePollingCardStatus, } } // parseInt64 解析字符串为 int64 func parseInt64(s string) int64 { var result int64 for _, c := range s { if c >= '0' && c <= '9' { result = result*10 + int64(c-'0') } } return result } // formatInt64 格式化 int64 为字符串 func formatInt64(n int64) string { if n == 0 { return "0" } var result []byte negative := n < 0 if negative { n = -n } for n > 0 { result = append([]byte{byte('0' + n%10)}, result...) n /= 10 } if negative { result = append([]byte{'-'}, result...) } return string(result) }