6a70713b30
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m0s
144 lines
5.2 KiB
Go
144 lines
5.2 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/polling"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// acquireConcurrencyScript 原子获取并发信号量的 Lua 脚本
|
||
// INCR + EXPIRE 合并为单个服务端操作，消除二者之间的崩溃窗口：
|
||
// 若 Worker 在 INCR 后、EXPIRE 前崩溃，key 将永久留在 Redis 导致计数器卡死。
|
||
// KEYS[1]: 当前并发计数 key
|
||
// ARGV[1]: 最大并发数；ARGV[2]: key TTL（秒）
|
||
// 返回 -1 表示超额拒绝，>0 表示成功获取后的计数值
|
||
var acquireConcurrencyScript = redis.NewScript(`
|
||
local current = redis.call('INCR', KEYS[1])
|
||
if tonumber(current) > tonumber(ARGV[1]) then
|
||
redis.call('DECR', KEYS[1])
|
||
return -1
|
||
end
|
||
redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
|
||
return current
|
||
`)
|
||
|
||
const cardTrafficSyncLockTTL = 10 * time.Minute
|
||
|
||
// PollingBase 轮询共享基类
|
||
// 封装并发控制、卡缓存、重入队、配置间隔查询等公共方法，所有 Handler 共享
|
||
type PollingBase struct {
|
||
redis *redis.Client
|
||
queueMgr *polling.PollingQueueManager
|
||
configMgr *polling.PollingConfigManager
|
||
iotCardStore *postgres.IotCardStore
|
||
logger *zap.Logger
|
||
verboseLog bool
|
||
}
|
||
|
||
// NewPollingBase 创建轮询共享基类
|
||
func NewPollingBase(
|
||
redisClient *redis.Client,
|
||
queueMgr *polling.PollingQueueManager,
|
||
configMgr *polling.PollingConfigManager,
|
||
iotCardStore *postgres.IotCardStore,
|
||
logger *zap.Logger,
|
||
verboseLog bool,
|
||
) *PollingBase {
|
||
return &PollingBase{
|
||
redis: redisClient,
|
||
queueMgr: queueMgr,
|
||
configMgr: configMgr,
|
||
iotCardStore: iotCardStore,
|
||
logger: logger,
|
||
verboseLog: verboseLog,
|
||
}
|
||
}
|
||
|
||
// acquireConcurrency 原子获取并发信号量
|
||
// 使用 Lua 脚本将 INCR 与 EXPIRE 合并为单个服务端操作，消除非原子窗口：
|
||
// 旧实现中若 Worker 在 INCR 后、EXPIRE 前崩溃，key 永久留在 Redis，计数器卡死。
|
||
func (b *PollingBase) acquireConcurrency(ctx context.Context, taskType string) bool {
|
||
shortType := shortTaskType(taskType)
|
||
configKey := constants.RedisPollingConcurrencyConfigKey(shortType)
|
||
currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType)
|
||
|
||
maxConcurrency, err := b.redis.Get(ctx, configKey).Int()
|
||
if err != nil {
|
||
maxConcurrency = constants.PollingDefaultMaxConcurrency
|
||
}
|
||
|
||
result, err := acquireConcurrencyScript.Run(
|
||
ctx, b.redis, []string{currentKey},
|
||
maxConcurrency, constants.PollingConcurrencyKeyTTL,
|
||
).Int64()
|
||
if err != nil {
|
||
b.logger.Warn("获取并发计数失败，放行任务", zap.Error(err))
|
||
return true
|
||
}
|
||
|
||
return result != -1
|
||
}
|
||
|
||
// releaseConcurrency 释放并发信号量
|
||
func (b *PollingBase) releaseConcurrency(ctx context.Context, taskType string) {
|
||
currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType)
|
||
if err := b.redis.Decr(ctx, currentKey).Err(); err != nil {
|
||
b.logger.Warn("释放并发计数失败", zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// 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 时调用此方法入队后再返回
|
||
func (b *PollingBase) requeueCard(ctx context.Context, cardID uint, taskType string) error {
|
||
card, err := b.getCardWithCache(ctx, cardID)
|
||
if err != nil {
|
||
b.logger.Warn("重入队：获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
// 获取卡失败时立即重入队（下次仍然尝试处理）
|
||
return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second))
|
||
}
|
||
|
||
intervals := b.configMgr.MergedTaskIntervals(card)
|
||
if len(intervals) == 0 {
|
||
// 兜底：配置未加载时延迟 30 秒重入队，防止卡从轮询永久消失
|
||
b.logger.Warn("无匹配轮询配置，30秒后重入队（配置可能未加载）",
|
||
zap.Uint("card_id", cardID), zap.String("task_type", taskType))
|
||
return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second))
|
||
}
|
||
info, ok := intervals[taskType]
|
||
if !ok || info.Interval <= 0 {
|
||
b.logger.Debug("轮询间隔为 NULL，不入队", zap.Uint("card_id", cardID), zap.String("task_type", taskType))
|
||
return nil
|
||
}
|
||
nextCheckAt := time.Now().Add(time.Duration(info.Interval) * time.Second)
|
||
return b.queueMgr.Requeue(ctx, cardID, taskType, nextCheckAt)
|
||
}
|