diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 48b2f2b..8df780c 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -264,7 +264,7 @@ func initServices(s *stores, deps *Dependencies) *services { Recharge: rechargeSvc.New(deps.DB, s.AssetWallet, s.AssetWalletTransaction, s.IotCard, s.Device, s.ShopSeriesAllocation, s.PackageSeries, s.CommissionRecord, wechatConfig, paymentLoader, deps.Logger), PollingConfig: pollingSvc.NewConfigService(s.PollingConfig, deps.Redis, deps.Logger), PollingConcurrency: pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis), - PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis), + PollingMonitoring: pollingSvc.NewMonitoringServiceWithQueueMgr(deps.Redis, pollingQueueMgr, deps.Logger), PollingAlert: pollingSvc.NewAlertService(s.PollingAlertRule, s.PollingAlertHistory, deps.Redis, deps.Logger), PollingCleanup: pollingSvc.NewCleanupService(s.DataCleanupConfig, s.DataCleanupLog, deps.Logger), PollingManualTrigger: pollingSvc.NewManualTriggerService(s.PollingManualTriggerLog, s.IotCard, deps.Redis, deps.Logger), diff --git a/internal/polling/queue_manager.go b/internal/polling/queue_manager.go index a69aace..a999b6e 100644 --- a/internal/polling/queue_manager.go +++ b/internal/polling/queue_manager.go @@ -186,3 +186,66 @@ func (m *PollingQueueManager) GetTotalQueueDepth(ctx context.Context, taskType s } return total, firstErr } + +// GetTotalDueCount 获取指定任务类型所有分片中已到期的队列数量。 +// 只统计 score <= now 的成员，用于后台监控展示真实待调度积压。 +func (m *PollingQueueManager) GetTotalDueCount(ctx context.Context, taskType string, now time.Time) (int64, error) { + var total int64 + var firstErr error + for i := 0; i < m.shardCount; i++ { + key := constants.RedisPollingShardQueueKey(i, taskType) + count, err := m.redis.ZCount(ctx, key, "-inf", fmt.Sprintf("%d", now.Unix())).Result() + if err != nil { + m.logger.Warn("获取分片到期数量失败", + zap.Int("shard_id", i), + zap.String("task_type", taskType), + zap.Error(err)) + if firstErr == nil { + firstErr = err + } + continue + } + total += count + } + return total, firstErr +} + +// GetAverageWaitTime 获取指定任务类型所有分片最早若干任务的平均等待秒数。 +// 每个分片最多取 samplePerShard 条，避免监控接口扫描完整队列。 +func (m *PollingQueueManager) GetAverageWaitTime(ctx context.Context, taskType string, now time.Time, samplePerShard int64) (float64, error) { + if samplePerShard <= 0 { + samplePerShard = 10 + } + + var totalWait float64 + var sampleCount int64 + var firstErr error + nowUnix := now.Unix() + + for i := 0; i < m.shardCount; i++ { + key := constants.RedisPollingShardQueueKey(i, taskType) + earliest, err := m.redis.ZRangeWithScores(ctx, key, 0, samplePerShard-1).Result() + if err != nil { + m.logger.Warn("获取分片最早任务失败", + zap.Int("shard_id", i), + zap.String("task_type", taskType), + zap.Error(err)) + if firstErr == nil { + firstErr = err + } + continue + } + for _, z := range earliest { + waitTime := float64(nowUnix) - z.Score + if waitTime > 0 { + totalWait += waitTime + } + sampleCount++ + } + } + + if sampleCount == 0 { + return 0, firstErr + } + return totalWait / float64(sampleCount), firstErr +} diff --git a/internal/service/polling/monitoring_service.go b/internal/service/polling/monitoring_service.go index 9ee5546..c16e681 100644 --- a/internal/service/polling/monitoring_service.go +++ b/internal/service/polling/monitoring_service.go @@ -114,14 +114,10 @@ func (s *MonitoringService) GetOverview(ctx context.Context) (*OverviewStats, er // GetQueueStatuses 获取所有队列状态 func (s *MonitoringService) GetQueueStatuses(ctx context.Context) ([]*QueueStatus, error) { - taskTypes := []string{ - constants.TaskTypePollingRealname, - constants.TaskTypePollingCarddata, - constants.TaskTypePollingPackage, - } + taskTypes := pollingMonitorTaskTypes() result := make([]*QueueStatus, 0, len(taskTypes)) - now := time.Now().Unix() + now := time.Now() for _, taskType := range taskTypes { status := &QueueStatus{ @@ -142,43 +138,43 @@ func (s *MonitoringService) GetQueueStatuses(ctx context.Context) ([]*QueueStatu 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()) } - // 获取到期数量（score <= now） - status.DueCount, _ = s.redis.ZCount(ctx, queueKey, "-inf", formatInt64(now)).Result() - // 获取手动触发队列待处理数 manualKey := constants.RedisPollingManualQueueKey(taskType) status.ManualPending, _ = s.redis.LLen(ctx, manualKey).Result() - // 计算平均等待时间（取最早的10个任务的平均等待时间） - earliest, err := s.redis.ZRangeWithScores(ctx, queueKey, 0, 9).Result() - if err == nil && len(earliest) > 0 { - var totalWait float64 - for _, z := range earliest { - waitTime := float64(now) - z.Score - if waitTime > 0 { - totalWait += waitTime - } - } - status.AvgWaitTime = totalWait / float64(len(earliest)) - } - 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 := []string{ - constants.TaskTypePollingRealname, - constants.TaskTypePollingCarddata, - constants.TaskTypePollingPackage, - } + taskTypes := pollingMonitorTaskTypes() result := make([]*TaskStats, 0, len(taskTypes)) @@ -269,11 +265,26 @@ func (s *MonitoringService) getTaskTypeName(taskType string) string { 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