csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

refactor: 流量系统重构 — 增量累加算法 + 日粒度缓冲 + 旧详单清理
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m16s
Details

Browse Source 
核心改造：
- 增量算法：流量计算从覆盖式改为增量累加（gateway - lastReading），支持上游运营商重置检测
- 日流量缓冲：insertDataUsageRecord 改为 Redis INCRBYFLOAT，每日凌晨落盘到 tb_card_daily_usage
- 运营商：新增 data_reset_day 字段（联通=27，其余=1）
- IoT卡：新增 last_gateway_reading_mb 字段存储上次网关读数
- 查询层：新建 TrafficQueryService 合并 Redis（今日）+ DB（历史）数据源
- 清理：删除 DataUsageRecord model/store，移除 polling_handler 旧引用

迁移：000094-000097（carrier字段、iot_card字段、数据初始化、日流量表）
This commit is contained in:
huang
2026-03-30 09:59:30 +08:00
parent f5dd2ce4ab
commit cebcada950
44 changed files with 1149 additions and 1613 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
internal/task/daily_traffic_flush.go Normal file
View File

@@ -0,0 +1,151 @@
package task
import (
"context"
"fmt"
"strconv"
"strings"
"time"
"github.com/hibiken/asynq"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
"github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
)
// DailyTrafficFlushHandler 每日流量落盘任务处理器
// 凌晨 2 点分批 SCAN 昨日 Redis key → UPSERT 到 tb_card_daily_usage → 删 key
type DailyTrafficFlushHandler struct {
redis *redis.Client
cardDailyUsageStore *postgres.CardDailyUsageStore
logger *zap.Logger
}
func NewDailyTrafficFlushHandler(
redisClient *redis.Client,
cardDailyUsageStore *postgres.CardDailyUsageStore,
logger *zap.Logger,
) *DailyTrafficFlushHandler {
return &DailyTrafficFlushHandler{
redis: redisClient,
cardDailyUsageStore: cardDailyUsageStore,
logger: logger,
}
}
func (h *DailyTrafficFlushHandler) HandleDailyTrafficFlush(ctx context.Context, _ *asynq.Task) error {
startTime := time.Now()
yesterday := time.Now().AddDate(0, 0, -1).Format("2006-01-02")
pattern := fmt.Sprintf("traffic:daily:*:%s", yesterday)
h.logger.Info("开始每日流量落盘",
zap.String("date", yesterday),
zap.String("pattern", pattern))
// 分批 SCAN 收集所有昨日 key
var allKeys []string
var cursor uint64
for {
keys, nextCursor, err := h.redis.Scan(ctx, cursor, pattern, 500).Result()
if err != nil {
h.logger.Error("SCAN Redis key 失败", zap.Error(err))
return err
}
allKeys = append(allKeys, keys...)
cursor = nextCursor
if cursor == 0 {
break
}
}
if len(allKeys) == 0 {
h.logger.Info("无需落盘，昨日无流量数据", zap.String("date", yesterday))
return nil
}
h.logger.Info("收集到待落盘 key",
zap.String("date", yesterday),
zap.Int("count", len(allKeys)))
yesterdayDate, _ := time.Parse("2006-01-02", yesterday)
// 分批处理：每批 200 条
flushedCount := 0
for i := 0; i < len(allKeys); i += 200 {
end := i + 200
if end > len(allKeys) {
end = len(allKeys)
}
batch := allKeys[i:end]
var records []*model.CardDailyUsage
var flushedKeys []string
for _, key := range batch {
cardID, err := parseCardIDFromKey(key)
if err != nil {
h.logger.Warn("解析 Redis key 失败", zap.String("key", key), zap.Error(err))
continue
}
val, err := h.redis.Get(ctx, key).Float64()
if err != nil {
h.logger.Warn("读取 Redis 值失败", zap.String("key", key), zap.Error(err))
continue
}
records = append(records, &model.CardDailyUsage{
IotCardID: cardID,
Date: yesterdayDate,
UsageMB: val,
})
flushedKeys = append(flushedKeys, key)
}
// 批量 UPSERT 到 DB
if len(records) > 0 {
if err := h.cardDailyUsageStore.UpsertBatch(ctx, records); err != nil {
h.logger.Error("批量 UPSERT 失败",
zap.Int("batch_size", len(records)),
zap.Error(err))
continue
}
}
// 批量删除已落盘的 Redis key
if len(flushedKeys) > 0 {
pipe := h.redis.Pipeline()
for _, key := range flushedKeys {
pipe.Del(ctx, key)
}
if _, err := pipe.Exec(ctx); err != nil {
h.logger.Warn("批量删除 Redis key 失败", zap.Error(err))
}
}
flushedCount += len(records)
}
h.logger.Info("每日流量落盘完成",
zap.String("date", yesterday),
zap.Int("flushed_count", flushedCount),
zap.Duration("duration", time.Since(startTime)))
return nil
}
// parseCardIDFromKey 从 Redis key "traffic:daily:{cardID}:{date}" 中解析 cardID
func parseCardIDFromKey(key string) (uint, error) {
parts := strings.Split(key, ":")
if len(parts) != 4 {
return 0, fmt.Errorf("invalid key format: %s", key)
}
id, err := strconv.ParseUint(parts[2], 10, 64)
if err != nil {
return 0, err
}
return uint(id), nil
}

194
internal/task/polling_handler.go
View File

@@ -35,7 +35,6 @@ type PollingHandler struct {
iotCardStore *postgres.IotCardStore
concurrencyStore *postgres.PollingConcurrencyConfigStore
deviceSimBindingStore *postgres.DeviceSimBindingStore
dataUsageRecordStore *postgres.DataUsageRecordStore
packageUsageStore *postgres.PackageUsageStore
usageService *packagepkg.UsageService
logger *zap.Logger
@@ -58,7 +57,6 @@ func NewPollingHandler(
iotCardStore: postgres.NewIotCardStore(db, redis),
concurrencyStore: postgres.NewPollingConcurrencyConfigStore(db),
deviceSimBindingStore: postgres.NewDeviceSimBindingStore(db, redis),
dataUsageRecordStore: postgres.NewDataUsageRecordStore(db),
packageUsageStore: postgres.NewPackageUsageStore(db, redis),
usageService: usageService,
logger: logger,
@@ -253,14 +251,10 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task)
zap.Uint64("card_id", cardID))
}
// 计算流量增量（处理跨月）
now := time.Now()
updates := h.calculateFlowUpdates(card, gatewayFlowMB, now)
updates, flowIncrementMB := h.calculateFlowUpdates(card, gatewayFlowMB, now)
updates["last_data_check_at"] = now
// 计算本次流量增量（用于套餐扣减）
flowIncrementMB := h.calculateFlowIncrement(card, gatewayFlowMB, now)
// 更新数据库
if err := h.db.Model(&model.IotCard{}).
Where("id = ?", cardID).
@@ -268,8 +262,7 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task)
h.logger.Error("更新卡流量信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
}
// 插入流量历史记录（异步，不阻塞主流程）
go h.insertDataUsageRecord(context.Background(), uint(cardID), gatewayFlowMB, card.CurrentMonthUsageMB, now, payload.IsManual)
go h.insertDataUsageRecord(context.Background(), uint(cardID), flowIncrementMB, now)
// 更新 Redis 缓存
h.updateCardCache(ctx, uint(cardID), map[string]any{
@@ -308,26 +301,28 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task)
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingCarddata)
}
// calculateFlowIncrement 任务 18.2: 计算本次流量增量
func (h *PollingHandler) calculateFlowIncrement(card *model.IotCard, gatewayFlowMB float64, now time.Time) float64 {
// 获取本月1号
currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location())
// 判断是否跨月
isCrossMonth := card.CurrentMonthStartDate == nil ||
card.CurrentMonthStartDate.Before(currentMonthStart)
if isCrossMonth {
// 跨月了：本月流量就是增量
return gatewayFlowMB
// getCarrierResetDay 读取运营商的上游流量重置日
func (h *PollingHandler) getCarrierResetDay(carrierID uint) int {
var carrier model.Carrier
if err := h.db.Select("data_reset_day").First(&carrier, carrierID).Error; err != nil {
return 1
}
// 同月内：计算增量
increment := gatewayFlowMB - card.CurrentMonthUsageMB
if increment < 0 {
return 0
if carrier.DataResetDay == 0 {
return 1
}
return increment
return carrier.DataResetDay
}
// isResetWindow 判断今天是否在运营商重置日窗口内
// 窗口 = 重置日当天 + 前一天（容错网关数据延迟）
func isResetWindow(now time.Time, resetDay int) bool {
today := now.Day()
if today == resetDay {
return true
}
resetDate := time.Date(now.Year(), now.Month(), resetDay, 0, 0, 0, 0, now.Location())
prevDay := resetDate.AddDate(0, 0, -1).Day()
return today == prevDay
}
// shouldStopCard 任务 18.4: 检查是否应该停机（所有套餐用完）
@@ -389,47 +384,64 @@ func (h *PollingHandler) stopCardByUsageExhausted(ctx context.Context, card *mod
zap.String("iccid", card.ICCID))
}
// calculateFlowUpdates 计算流量更新值（处理跨月逻辑）
func (h *PollingHandler) calculateFlowUpdates(card *model.IotCard, gatewayFlowMB float64, now time.Time) map[string]any {
// calculateFlowUpdates 计算流量更新值和增量（合并原 calculateFlowUpdates 和 calculateFlowIncrement）
// 增量算法：increment = 当前网关读数 - 上次网关读数，检测上游重置和跨自然月
func (h *PollingHandler) calculateFlowUpdates(card *model.IotCard, gatewayFlowMB float64, now time.Time) (map[string]any, float64) {
updates := make(map[string]any)
// 获取本月1号
currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location())
// 增量 = 当前读数 - 上次读数
increment := gatewayFlowMB - card.LastGatewayReadingMB
// 判断是否跨月
if increment < 0 {
// 当前值比上次小，检查是否为上游运营商正常重置
resetDay := h.getCarrierResetDay(card.CarrierID)
if isResetWindow(now, resetDay) {
// 在重置日窗口内，视为正常上游重置，本次原始值即为增量
increment = gatewayFlowMB
h.logger.Info("检测到上游运营商重置",
zap.Uint("card_id", card.ID),
zap.Float64("gateway_flow", gatewayFlowMB),
zap.Float64("last_reading", card.LastGatewayReadingMB),
zap.Int("reset_day", resetDay))
} else {
// 非重置日出现值下降，异常，不计入
h.logger.Warn("流量异常：非重置日出现值下降",
zap.Uint("card_id", card.ID),
zap.Float64("gateway_flow", gatewayFlowMB),
zap.Float64("last_reading", card.LastGatewayReadingMB))
increment = 0
}
}
// 始终更新网关读数
updates["last_gateway_reading_mb"] = gatewayFlowMB
// 检测跨自然月
currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location())
isCrossMonth := card.CurrentMonthStartDate == nil ||
card.CurrentMonthStartDate.Before(currentMonthStart)
if isCrossMonth {
// 跨月了：保存上月总量，重置本月
h.logger.Info("检测到跨月，重置流量计数",
zap.Uint("card_id", card.ID),
zap.Float64("last_month_total", card.CurrentMonthUsageMB),
zap.Float64("new_month_usage", gatewayFlowMB))
// 计算本次增量：上月最后值 + 本月当前值
increment := card.CurrentMonthUsageMB + gatewayFlowMB
zap.Float64("last_month_total", card.CurrentMonthUsageMB))
updates["last_month_total_mb"] = card.CurrentMonthUsageMB
updates["current_month_start_date"] = currentMonthStart
updates["current_month_usage_mb"] = gatewayFlowMB
updates["data_usage_mb"] = card.DataUsageMB + int64(increment)
} else {
// 同月内：计算增量
increment := gatewayFlowMB - card.CurrentMonthUsageMB
if increment < 0 {
// Gateway 返回值比记录的小，可能是数据异常，不更新
h.logger.Warn("流量异常：Gateway返回值小于记录值",
zap.Uint("card_id", card.ID),
zap.Float64("gateway_flow", gatewayFlowMB),
zap.Float64("recorded_flow", card.CurrentMonthUsageMB))
increment = 0
}
updates["current_month_usage_mb"] = gatewayFlowMB
// 跨月重置为 0，再加上本次增量
if increment > 0 {
updates["data_usage_mb"] = card.DataUsageMB + int64(increment)
updates["current_month_usage_mb"] = increment
} else {
updates["current_month_usage_mb"] = float64(0)
}
} else if increment > 0 {
// 同月内，增量累加（使用 gorm.Expr 保证原子性）
updates["current_month_usage_mb"] = gorm.Expr("current_month_usage_mb + ?", increment)
}
// 更新卡生命周期总用量
if increment > 0 {
updates["data_usage_mb"] = gorm.Expr("data_usage_mb + ?", int64(increment))
}
// 首次流量查询初始化
@@ -437,7 +449,7 @@ func (h *PollingHandler) calculateFlowUpdates(card *model.IotCard, gatewayFlowMB
updates["current_month_start_date"] = currentMonthStart
}
return updates
return updates, increment
}
// HandlePackageCheck 处理套餐检查任务
@@ -813,44 +825,24 @@ func (h *PollingHandler) updateStats(ctx context.Context, taskType string, succe
_, _ = pipe.Exec(ctx)
}
// insertDataUsageRecord 插入流量历史记录
func (h *PollingHandler) insertDataUsageRecord(ctx context.Context, cardID uint, currentUsageMB, previousUsageMB float64, checkTime time.Time, isManual bool) {
// 计算流量增量
var dataIncreaseMB int64
if currentUsageMB > previousUsageMB {
dataIncreaseMB = int64(currentUsageMB - previousUsageMB)
// insertDataUsageRecord 写入日流量缓冲到 Redis
// 增量 <= 0 时直接跳过；增量 > 0 时 INCRBYFLOAT 写 Redis + 48h TTL
func (h *PollingHandler) insertDataUsageRecord(ctx context.Context, cardID uint, incrementMB float64, checkTime time.Time) {
if incrementMB <= 0 {
return
}
// 确定数据来源
source := "polling"
if isManual {
source = "manual"
}
dateStr := checkTime.Format("2006-01-02")
key := constants.RedisCardDailyTrafficKey(cardID, dateStr)
// 创建流量记录
record := &model.DataUsageRecord{
IotCardID: cardID,
DataUsageMB: int64(currentUsageMB),
DataIncreaseMB: dataIncreaseMB,
CheckTime: checkTime,
Source: source,
}
// 插入记录
if err := h.dataUsageRecordStore.Create(ctx, record); err != nil {
h.logger.Warn("插入流量历史记录失败",
if err := h.redis.IncrByFloat(ctx, key, incrementMB).Err(); err != nil {
h.logger.Warn("写入日流量缓冲失败",
zap.Uint("card_id", cardID),
zap.Int64("data_usage_mb", record.DataUsageMB),
zap.Int64("data_increase_mb", record.DataIncreaseMB),
zap.String("source", source),
zap.Float64("increment_mb", incrementMB),
zap.Error(err))
} else {
h.logger.Debug("流量历史记录已插入",
zap.Uint("card_id", cardID),
zap.Int64("data_usage_mb", record.DataUsageMB),
zap.Int64("data_increase_mb", record.DataIncreaseMB),
zap.String("source", source))
return
}
h.redis.Expire(ctx, key, 48*time.Hour)
}
// updateCardCache 更新卡缓存
@@ -910,6 +902,16 @@ func (h *PollingHandler) getCardWithCache(ctx context.Context, cardID uint) (*mo
card.CurrentMonthUsageMB = usage
}
}
if v, ok := result["last_gateway_reading_mb"]; ok {
if reading, err := strconv.ParseFloat(v, 64); err == nil {
card.LastGatewayReadingMB = reading
}
}
if v, ok := result["data_usage_mb"]; ok {
if usage, err := strconv.ParseInt(v, 10, 64); err == nil {
card.DataUsageMB = usage
}
}
if v, ok := result["series_id"]; ok {
if id, err := strconv.ParseUint(v, 10, 64); err == nil {
seriesID := uint(id)
@@ -930,14 +932,16 @@ func (h *PollingHandler) getCardWithCache(ctx context.Context, cardID uint) (*mo
go func() {
cacheCtx := context.Background()
cacheData := map[string]any{
"id": card.ID,
"iccid": card.ICCID,
"card_category": card.CardCategory,
"real_name_status": card.RealNameStatus,
"network_status": card.NetworkStatus,
"carrier_id": card.CarrierID,
"current_month_usage_mb": card.CurrentMonthUsageMB,
"cached_at": time.Now().Unix(),
"id": card.ID,
"iccid": card.ICCID,
"card_category": card.CardCategory,
"real_name_status": card.RealNameStatus,
"network_status": card.NetworkStatus,
"carrier_id": card.CarrierID,
"current_month_usage_mb": card.CurrentMonthUsageMB,
"last_gateway_reading_mb": card.LastGatewayReadingMB,
"data_usage_mb": card.DataUsageMB,
"cached_at": time.Now().Unix(),
}
if card.SeriesID != nil {
cacheData["series_id"] = *card.SeriesID