From 6a70713b301f5dd4afa4e402727bed5420f2761a Mon Sep 17 00:00:00 2001 From: Break Date: Tue, 9 Jun 2026 10:54:13 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=A0=E5=88=86=E5=B8=83=E5=BC=8F=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/service/iot_card/service.go | 34 ++++++++++++++++++++++ internal/service/package/usage_service.go | 4 ++- internal/task/polling_base.go | 24 ++++++++++++++++ internal/task/polling_carddata_handler.go | 35 +++++++++++++---------- pkg/constants/redis.go | 7 +++++ 5 files changed, 88 insertions(+), 16 deletions(-) diff --git a/internal/service/iot_card/service.go b/internal/service/iot_card/service.go index 2487589..f5c3c0d 100644 --- a/internal/service/iot_card/service.go +++ b/internal/service/iot_card/service.go @@ -20,6 +20,8 @@ import ( "gorm.io/gorm" ) +const cardTrafficSyncLockTTL = 10 * time.Minute + // PollingCallback 轮询回调接口 // 用于在卡生命周期事件发生时通知轮询调度器 type PollingCallback interface { @@ -135,6 +137,24 @@ func (s *Service) SetRedisClient(redisClient *redis.Client) { s.redis = redisClient } +// acquireCardTrafficSyncLock 获取卡流量同步锁，避免主动刷新与轮询重复统计同一上游读数。 +func (s *Service) acquireCardTrafficSyncLock(ctx context.Context, cardID uint) (bool, error) { + if s.redis == nil { + return true, nil + } + return s.redis.SetNX(ctx, constants.RedisCardTrafficSyncLockKey(cardID), "1", cardTrafficSyncLockTTL).Result() +} + +// releaseCardTrafficSyncLock 释放卡流量同步锁。 +func (s *Service) releaseCardTrafficSyncLock(ctx context.Context, cardID uint) { + if s.redis == nil { + return + } + if err := s.redis.Del(ctx, constants.RedisCardTrafficSyncLockKey(cardID)).Err(); err != nil { + s.logger.Warn("释放卡流量同步锁失败", zap.Uint("card_id", cardID), zap.Error(err)) + } +} + func (s *Service) ListStandalone(ctx context.Context, req *dto.ListStandaloneIotCardRequest) (*dto.ListStandaloneIotCardResponse, error) { page := req.Page pageSize := req.PageSize @@ -1418,6 +1438,20 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) return err } + locked, lockErr := s.acquireCardTrafficSyncLock(ctx, card.ID) + if lockErr != nil { + return errors.Wrap(errors.CodeInternalError, lockErr, "获取卡流量同步锁失败") + } else if !locked { + return errors.New(errors.CodeTooManyRequests, "卡流量正在同步，请稍后重试") + } else { + defer s.releaseCardTrafficSyncLock(ctx, card.ID) + latestCard, loadErr := s.iotCardStore.GetByID(ctx, card.ID) + if loadErr != nil { + return errors.Wrap(errors.CodeInternalError, loadErr, "刷新卡数据失败") + } + card = latestCard + } + syncTime := time.Now() updates := map[string]any{ "last_sync_time": syncTime, diff --git a/internal/service/package/usage_service.go b/internal/service/package/usage_service.go index 5cedfd0..a00d1db 100644 --- a/internal/service/package/usage_service.go +++ b/internal/service/package/usage_service.go @@ -13,6 +13,7 @@ import ( "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" + "gorm.io/gorm/clause" ) const ( @@ -301,7 +302,8 @@ func (s *UsageService) queryActivePackages(tx *gorm.DB, carrierType string, carr var packages []*model.PackageUsage if err := query. - Order("CASE WHEN master_usage_id IS NOT NULL THEN 0 ELSE 1 END, priority ASC"). + Clauses(clause.Locking{Strength: "UPDATE"}). + Order("CASE WHEN master_usage_id IS NOT NULL THEN 0 ELSE 1 END, priority ASC, id ASC"). Find(&packages).Error; err != nil { return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询生效套餐失败") } diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 7a5eaf9..4da8b4a 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -28,6 +28,8 @@ var acquireConcurrencyScript = redis.NewScript(` return current `) +const cardTrafficSyncLockTTL = 10 * time.Minute + // PollingBase 轮询共享基类 // 封装并发控制、卡缓存、重入队、配置间隔查询等公共方法，所有 Handler 共享 type PollingBase struct { @@ -91,6 +93,28 @@ func (b *PollingBase) releaseConcurrency(ctx context.Context, taskType string) { } } +// acquireCardTrafficSyncLock 获取卡流量同步锁，避免轮询与手动刷新重复统计同一上游读数。 +func (b *PollingBase) acquireCardTrafficSyncLock(ctx context.Context, cardID uint) (bool, error) { + if b.redis == nil { + return true, nil + } + locked, err := b.redis.SetNX(ctx, constants.RedisCardTrafficSyncLockKey(cardID), "1", cardTrafficSyncLockTTL).Result() + if err != nil { + return false, err + } + return locked, nil +} + +// releaseCardTrafficSyncLock 释放卡流量同步锁。 +func (b *PollingBase) releaseCardTrafficSyncLock(ctx context.Context, cardID uint) { + if b.redis == nil { + return + } + if err := b.redis.Del(ctx, constants.RedisCardTrafficSyncLockKey(cardID)).Err(); err != nil { + b.logger.Warn("释放卡流量同步锁失败", zap.Uint("card_id", cardID), zap.Error(err)) + } +} + // requeueCard 将卡按匹配配置间隔重新入队分片 Sorted Set // ⚠️ 关键：Lua 脚本原子出队后卡已从队列移除，若并发满时直接 return 会导致卡永久丢失 // 调用方必须在 acquireConcurrency 返回 false 时调用此方法入队后再返回 diff --git a/internal/task/polling_carddata_handler.go b/internal/task/polling_carddata_handler.go index c50eceb..dc04930 100644 --- a/internal/task/polling_carddata_handler.go +++ b/internal/task/polling_carddata_handler.go @@ -62,7 +62,19 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, t *asynq.Task) erro } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingCarddata) - card, err := h.base.getCardWithCache(ctx, cardID) + locked, lockErr := h.base.acquireCardTrafficSyncLock(ctx, cardID) + if lockErr != nil { + h.base.logger.Warn("获取卡流量同步锁失败，延迟重入队", zap.Uint("card_id", cardID), zap.Error(lockErr)) + return h.base.queueMgr.Requeue(ctx, cardID, constants.TaskTypePollingCarddata, time.Now().Add(5*time.Second)) + } + if !locked { + h.base.logger.Info("卡流量同步正在执行，延迟重入队", zap.Uint("card_id", cardID)) + return h.base.queueMgr.Requeue(ctx, cardID, constants.TaskTypePollingCarddata, time.Now().Add(5*time.Second)) + } + defer h.base.releaseCardTrafficSyncLock(ctx, cardID) + + h.base.invalidateCardCache(ctx, cardID) + card, err := h.iotCardStore.GetByID(ctx, cardID) if err != nil { if isNotFound(err) { return nil @@ -105,25 +117,18 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, t *asynq.Task) erro if updateErr := h.iotCardStore.UpdateFields(ctx, cardID, updates); updateErr != nil { h.base.logger.Error("更新卡流量信息失败", zap.Uint("card_id", cardID), zap.Error(updateErr)) + h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startTime)) + return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) } go h.base.insertDataUsageRecord(context.Background(), cardID, flowIncrementMB, now) - cacheUpdates := map[string]any{} - if reading, ok := updates["last_gateway_reading_mb"]; ok { - cacheUpdates["last_gateway_reading_mb"] = reading + if refreshedCard, loadErr := h.iotCardStore.GetByID(ctx, cardID); loadErr == nil { + writeCardToCache(h.base, constants.RedisPollingCardInfoKey(cardID), refreshedCard) + } else { + h.base.invalidateCardCache(ctx, cardID) + h.base.logger.Warn("刷新卡缓存失败", zap.Uint("card_id", cardID), zap.Error(loadErr)) } - if isCrossMonth { - // 跨月时同步更新缓存中的本月开始日期和本月用量，避免下次从缓存读取时再次误判跨月 - currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location()) - cacheUpdates["current_month_start_date"] = currentMonthStart.Unix() - if flowIncrementMB > 0 { - cacheUpdates["current_month_usage_mb"] = flowIncrementMB - } else { - cacheUpdates["current_month_usage_mb"] = float64(0) - } - } - h.base.updateCardCache(ctx, cardID, cacheUpdates) if flowIncrementMB > 0 && h.usageService != nil { if deductErr := h.usageService.DeductDataUsage(ctx, constants.AssetTypeIotCard, cardID, flowIncrementMB); deductErr != nil { diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index 3bae739..c745e58 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -449,6 +449,13 @@ func RedisPackageUsageRemainderKey(carrierType string, carrierID uint) string { return fmt.Sprintf("package:usage:remainder:%s:%d", carrierType, carrierID) } +// RedisCardTrafficSyncLockKey 生成卡流量同步锁 Redis 键 +// 用途：串行化轮询与手动刷新，避免同一上游读数被重复扣减套餐流量 +// 过期时间：10 分钟 +func RedisCardTrafficSyncLockKey(cardID uint) string { + return fmt.Sprintf("traffic:sync:lock:card:%d", cardID) +} + // ======================================== // 轮询分片队列相关 Redis Key // ========================================