监控

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),

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
+}

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{
+	taskTypes := pollingMonitorTaskTypes()
-		constants.TaskTypePollingRealname,
-		constants.TaskTypePollingCarddata,
-		constants.TaskTypePollingPackage,
-	}
 
 	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{
+	taskTypes := pollingMonitorTaskTypes()
-		constants.TaskTypePollingRealname,
-		constants.TaskTypePollingCarddata,
-		constants.TaskTypePollingPackage,
-	}
 
 	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