diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 3d11030d..064f5ecb 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -87,6 +87,12 @@ func (b *PollingBase) requeueRateLimited(cardID uint, taskType string) error { delay = configured } } + ctx, cancel := pollingFallbackContext() + card, err := b.getCardWithCache(ctx, cardID) + cancel() + if err == nil && !card.EnablePolling { + return nil + } return b.requeueCardAt(cardID, taskType, time.Now().Add(delay)) } @@ -189,6 +195,10 @@ func (b *PollingBase) ensureMissingTask(ctx context.Context, cardID uint, taskTy return err } + if !card.EnablePolling { + // 已关闭轮询的卡不补齐任务,否则卡状态与套餐任务会互相补回队列,关闭轮询无法生效。 + return nil + } info, ok := b.configMgr.MergedTaskIntervals(card)[taskType] if !ok || info.Interval <= 0 { return nil @@ -231,6 +241,11 @@ func (b *PollingBase) pollingNextCheckAt(ctx context.Context, cardID uint, taskT zap.Uint("card_id", cardID), zap.String("task_type", taskType), zap.Error(err)) return time.Now().Add(pollingRequeueFallbackInterval), true } + if !card.EnablePolling { + // 执行中途被关闭轮询的卡不再重入队(关闭时已移出队列并清除缓存,此处读到的是库中最新值)。 + b.logger.Info("卡已关闭轮询,不再重入队", zap.Uint("card_id", cardID), zap.String("task_type", taskType)) + return time.Time{}, false + } intervals := b.configMgr.MergedTaskIntervals(card) if len(intervals) == 0 { // 兜底:配置未加载时延迟 30 秒重入队,防止卡从轮询永久消失 diff --git a/internal/task/polling_card_cache.go b/internal/task/polling_card_cache.go index 1dc3394e..9c670088 100644 --- a/internal/task/polling_card_cache.go +++ b/internal/task/polling_card_cache.go @@ -18,8 +18,10 @@ func (b *PollingBase) getCardWithCache(ctx context.Context, cardID uint) (*model result, err := b.redis.HGetAll(ctx, key).Result() if err == nil && len(result) > 0 { card := parseCardFromCache(cardID, result) - if _, ok := result["asset_status"]; !ok { - b.backfillAssetStatus(ctx, key, card) + _, hasAssetStatus := result["asset_status"] + _, hasEnablePolling := result["enable_polling"] + if !hasAssetStatus || !hasEnablePolling { + b.backfillCardFields(ctx, key, card) } return card, nil } @@ -36,16 +38,20 @@ func (b *PollingBase) getCardWithCache(ctx context.Context, cardID uint) (*model return card, nil } -// backfillAssetStatus 为分层轮询补齐旧缓存缺少的资产状态:只读取并回写这一个字段, -// 不整条重建缓存,避免覆盖缓存中单独维护的读数等字段。 -func (b *PollingBase) backfillAssetStatus(ctx context.Context, key string, card *model.IotCard) { - var statuses []int - if err := b.db.WithContext(ctx).Model(&model.IotCard{}).Where("id = ?", card.ID).Limit(1).Pluck("asset_status", &statuses).Error; err != nil || len(statuses) == 0 { +// backfillCardFields 为旧缓存补齐资产状态与轮询开关:只读取并回写这两个字段, +// 不整条重建缓存,避免覆盖缓存中单独维护的读数等字段。读取失败时保留缓存解析结果(轮询开关按开启)。 +func (b *PollingBase) backfillCardFields(ctx context.Context, key string, card *model.IotCard) { + var rows []struct { + AssetStatus int + EnablePolling bool + } + if err := b.db.WithContext(ctx).Model(&model.IotCard{}).Select("asset_status, enable_polling"). + Where("id = ?", card.ID).Limit(1).Scan(&rows).Error; err != nil || len(rows) == 0 { return } - card.AssetStatus = statuses[0] - if err := b.redis.HSet(ctx, key, "asset_status", card.AssetStatus).Err(); err != nil { - b.logger.Debug("回写卡缓存资产状态失败", zap.Uint("card_id", card.ID), zap.Error(err)) + card.AssetStatus, card.EnablePolling = rows[0].AssetStatus, rows[0].EnablePolling + if err := b.redis.HSet(ctx, key, "asset_status", card.AssetStatus, "enable_polling", boolToStr(card.EnablePolling)).Err(); err != nil { + b.logger.Debug("回写卡缓存字段失败", zap.Uint("card_id", card.ID), zap.Error(err)) } } @@ -122,6 +128,11 @@ func parseCardFromCache(cardID uint, result map[string]string) *model.IotCard { card.AssetStatus = status } } + // 轮询开关:所有关闭/开启路径都会清除卡缓存,缓存中的取值与库一致;旧缓存缺字段时按开启并由读取方回填。 + card.EnablePolling = true + if v, ok := result["enable_polling"]; ok { + card.EnablePolling = v == "1" || v == "true" + } if v, ok := result["is_standalone"]; ok { card.IsStandalone = v == "1" || v == "true" } else { @@ -158,6 +169,7 @@ func writeCardToCache(b *PollingBase, key string, card *model.IotCard) { "data_usage_mb": card.DataUsageMB, "stop_reason": card.StopReason, "is_standalone": boolToStr(card.IsStandalone), + "enable_polling": boolToStr(card.EnablePolling), "asset_status": card.AssetStatus, "cached_at": time.Now().Unix(), }