fix: 修正轮询配置多匹配逻辑,支持同卡匹配多个配置并按 priority 合并 interval
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m27s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m27s
核心变更: - MatchConfig 改为 MatchConfigs,返回所有匹配配置 - MergedTaskIntervals 按 task type 合并各配置,选取最高优先级(非 nil 且最小 priority 值) - hasAnyEnabledInterval 过滤所有 interval 均为 NULL 的配置 - calcInitialDelay 重构为纯函数,接收 interval 参数 - 移除 getEnabledTaskTypes 和 getIntervalByTaskType(被 MergedTaskIntervals 替代) - scheduler.go 新增心跳 key + 顶层 panic recovery + Init 完成守卫 - initializer.go 批量失败日志升级为 Error,逐条检查 Pipeline 命令错误 - 数据迁移:禁用 id=29 的轮询配置(所有 interval 均为 NULL) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -27,6 +27,12 @@ type PollingConfigManager struct {
|
||||
refreshOnce sync.Once
|
||||
}
|
||||
|
||||
// TaskTypeInterval 某任务类型的轮询间隔信息
|
||||
type TaskTypeInterval struct {
|
||||
Interval int // 秒
|
||||
Priority int // 来源配置的 priority
|
||||
}
|
||||
|
||||
// NewPollingConfigManager 创建配置管理器
|
||||
func NewPollingConfigManager(configStore *postgres.PollingConfigStore, redisClient *redis.Client, logger *zap.Logger) *PollingConfigManager {
|
||||
return &PollingConfigManager{
|
||||
@@ -81,16 +87,79 @@ func (m *PollingConfigManager) Start(ctx context.Context) {
|
||||
|
||||
// MatchConfig 按优先级匹配第一个符合条件的轮询配置
|
||||
// 配置按 priority ASC 排序(DB 层保证),数字越小优先级越高
|
||||
// MatchConfigs()[0] 的前向兼容包装
|
||||
func (m *PollingConfigManager) MatchConfig(card *model.IotCard) *model.PollingConfig {
|
||||
configs := m.MatchConfigs(card)
|
||||
if len(configs) == 0 {
|
||||
return nil
|
||||
}
|
||||
return configs[0]
|
||||
}
|
||||
|
||||
// MatchConfigs 按 priority ASC 遍历所有启用的配置,返回所有满足匹配条件且至少有一个非 NULL interval 的配置
|
||||
// 无匹配时返回空切片
|
||||
func (m *PollingConfigManager) MatchConfigs(card *model.IotCard) []*model.PollingConfig {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
var result []*model.PollingConfig
|
||||
for _, cfg := range m.configs {
|
||||
if matchConfigConditions(cfg, card) {
|
||||
return cfg
|
||||
if matchConfigConditions(cfg, card) && hasAnyEnabledInterval(cfg) {
|
||||
result = append(result, cfg)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return result
|
||||
}
|
||||
|
||||
// hasAnyEnabledInterval 检查配置是否有至少一个非 NULL 且大于 0 的轮询间隔
|
||||
func hasAnyEnabledInterval(cfg *model.PollingConfig) bool {
|
||||
return (cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0) ||
|
||||
(cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0) ||
|
||||
(cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0) ||
|
||||
(cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0) ||
|
||||
(cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0)
|
||||
}
|
||||
|
||||
// MergedTaskIntervals 对所有匹配配置按 priority 遍历,对每种 task type 选取最高优先级的非 nil interval
|
||||
// 返回 map: taskType → TaskTypeInterval{Interval, Priority}
|
||||
func (m *PollingConfigManager) MergedTaskIntervals(card *model.IotCard) map[string]TaskTypeInterval {
|
||||
configs := m.MatchConfigs(card)
|
||||
merged := make(map[string]TaskTypeInterval)
|
||||
|
||||
for _, cfg := range configs {
|
||||
priority := int(cfg.Priority)
|
||||
if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 {
|
||||
taskType := constants.TaskTypePollingRealname
|
||||
if _, exists := merged[taskType]; !exists {
|
||||
merged[taskType] = TaskTypeInterval{Interval: *cfg.RealnameCheckInterval, Priority: priority}
|
||||
}
|
||||
}
|
||||
if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 {
|
||||
taskType := constants.TaskTypePollingCarddata
|
||||
if _, exists := merged[taskType]; !exists {
|
||||
merged[taskType] = TaskTypeInterval{Interval: *cfg.CarddataCheckInterval, Priority: priority}
|
||||
}
|
||||
}
|
||||
if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 {
|
||||
taskType := constants.TaskTypePollingPackage
|
||||
if _, exists := merged[taskType]; !exists {
|
||||
merged[taskType] = TaskTypeInterval{Interval: *cfg.PackageCheckInterval, Priority: priority}
|
||||
}
|
||||
}
|
||||
if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 {
|
||||
taskType := constants.TaskTypePollingProtect
|
||||
if _, exists := merged[taskType]; !exists {
|
||||
merged[taskType] = TaskTypeInterval{Interval: *cfg.ProtectCheckInterval, Priority: priority}
|
||||
}
|
||||
}
|
||||
if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 {
|
||||
taskType := constants.TaskTypePollingCardStatus
|
||||
if _, exists := merged[taskType]; !exists {
|
||||
merged[taskType] = TaskTypeInterval{Interval: *cfg.CardStatusCheckInterval, Priority: priority}
|
||||
}
|
||||
}
|
||||
}
|
||||
return merged
|
||||
}
|
||||
|
||||
// matchConfigConditions 检查卡是否满足配置的匹配条件
|
||||
|
||||
@@ -157,7 +157,8 @@ func (p *PollingInitializer) run(ctx context.Context) {
|
||||
}
|
||||
|
||||
if initErr := p.initBatch(ctx, cards); initErr != nil {
|
||||
p.logger.Warn("批量初始化失败", zap.Error(initErr))
|
||||
p.logger.Error("批量初始化失败,该批次卡未入轮询队列",
|
||||
zap.Uint("batch_start_id", lastID), zap.Error(initErr))
|
||||
}
|
||||
|
||||
lastID = cards[len(cards)-1].ID
|
||||
@@ -206,16 +207,24 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa
|
||||
if cmdCount == 0 {
|
||||
return
|
||||
}
|
||||
if _, execErr := pipe.Exec(ctx); execErr != nil {
|
||||
p.logger.Warn("Pipeline flush 失败,继续下一批", zap.Error(execErr))
|
||||
cmds, execErr := pipe.Exec(ctx)
|
||||
if execErr != nil {
|
||||
p.logger.Error("Pipeline flush 失败,部分卡可能未入队", zap.Error(execErr))
|
||||
}
|
||||
// 逐条检查,记录具体失败命令(不中断批次)
|
||||
for _, cmd := range cmds {
|
||||
if cmdErr := cmd.Err(); cmdErr != nil && cmdErr != redis.Nil {
|
||||
p.logger.Warn("Pipeline 单条命令失败",
|
||||
zap.String("cmd", cmd.Name()), zap.Error(cmdErr))
|
||||
}
|
||||
}
|
||||
pipe = p.redis.Pipeline()
|
||||
cmdCount = 0
|
||||
}
|
||||
|
||||
for _, card := range cards {
|
||||
cfg := p.configMgr.MatchConfig(card)
|
||||
if cfg == nil {
|
||||
intervals := p.configMgr.MergedTaskIntervals(card)
|
||||
if len(intervals) == 0 {
|
||||
skippedCards++
|
||||
continue
|
||||
}
|
||||
@@ -224,37 +233,10 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa
|
||||
shardID := int(card.ID) % p.queueMgr.shardCount
|
||||
cardIDStr := fmt.Sprintf("%d", card.ID)
|
||||
|
||||
if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 {
|
||||
nextCheck := calculateNextCheckTime(card.LastRealNameCheckAt, *cfg.RealnameCheckInterval, now)
|
||||
pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingRealname), redis.Z{
|
||||
Score: float64(nextCheck.Unix()), Member: cardIDStr,
|
||||
})
|
||||
cmdCount++
|
||||
}
|
||||
if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 {
|
||||
nextCheck := calculateNextCheckTime(card.LastDataCheckAt, *cfg.CarddataCheckInterval, now)
|
||||
pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingCarddata), redis.Z{
|
||||
Score: float64(nextCheck.Unix()), Member: cardIDStr,
|
||||
})
|
||||
cmdCount++
|
||||
}
|
||||
if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 {
|
||||
nextCheck := calculateNextCheckTime(card.LastDataCheckAt, *cfg.PackageCheckInterval, now)
|
||||
pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingPackage), redis.Z{
|
||||
Score: float64(nextCheck.Unix()), Member: cardIDStr,
|
||||
})
|
||||
cmdCount++
|
||||
}
|
||||
if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 {
|
||||
nextCheck := calculateNextCheckTime(card.LastProtectCheckAt, *cfg.ProtectCheckInterval, now)
|
||||
pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingProtect), redis.Z{
|
||||
Score: float64(nextCheck.Unix()), Member: cardIDStr,
|
||||
})
|
||||
cmdCount++
|
||||
}
|
||||
if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 {
|
||||
nextCheck := calculateNextCheckTime(card.LastCardStatusCheckAt, *cfg.CardStatusCheckInterval, now)
|
||||
pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingCardStatus), redis.Z{
|
||||
for taskType, info := range intervals {
|
||||
lastCheckAt := lastCheckAtByTaskType(card, taskType)
|
||||
nextCheck := calculateNextCheckTime(lastCheckAt, info.Interval, now)
|
||||
pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, taskType), redis.Z{
|
||||
Score: float64(nextCheck.Unix()), Member: cardIDStr,
|
||||
})
|
||||
cmdCount++
|
||||
@@ -294,6 +276,22 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa
|
||||
return nil
|
||||
}
|
||||
|
||||
// lastCheckAtByTaskType 根据 task type 返回对应的上次检查时间字段
|
||||
func lastCheckAtByTaskType(card *model.IotCard, taskType string) *time.Time {
|
||||
switch taskType {
|
||||
case constants.TaskTypePollingRealname:
|
||||
return card.LastRealNameCheckAt
|
||||
case constants.TaskTypePollingCarddata, constants.TaskTypePollingPackage:
|
||||
return card.LastDataCheckAt
|
||||
case constants.TaskTypePollingProtect:
|
||||
return card.LastProtectCheckAt
|
||||
case constants.TaskTypePollingCardStatus:
|
||||
return card.LastCardStatusCheckAt
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// calculateNextCheckTime 计算下次检查时间
|
||||
func calculateNextCheckTime(lastCheckAt *time.Time, intervalSeconds int, now time.Time) time.Time {
|
||||
if lastCheckAt == nil {
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
)
|
||||
|
||||
// PollingLifecycleService 卡生命周期轮询管理服务
|
||||
@@ -134,68 +133,23 @@ func (s *PollingLifecycleService) shouldEnqueue(ctx context.Context, card *model
|
||||
|
||||
// enqueueCard 按匹配配置将卡入队分片队列
|
||||
func (s *PollingLifecycleService) enqueueCard(ctx context.Context, card *model.IotCard) {
|
||||
cfg := s.configMgr.MatchConfig(card)
|
||||
if cfg == nil {
|
||||
intervals := s.configMgr.MergedTaskIntervals(card)
|
||||
if len(intervals) == 0 {
|
||||
return
|
||||
}
|
||||
taskTypes := getEnabledTaskTypes(cfg)
|
||||
for _, taskType := range taskTypes {
|
||||
if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(card, cfg, taskType)); err != nil {
|
||||
for taskType, info := range intervals {
|
||||
if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(info.Interval)); err != nil {
|
||||
s.logger.Warn("卡入队失败",
|
||||
zap.Uint("card_id", card.ID), zap.String("task_type", taskType), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// getEnabledTaskTypes 返回配置中启用的任务类型列表
|
||||
func getEnabledTaskTypes(cfg *model.PollingConfig) []string {
|
||||
var types []string
|
||||
if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 {
|
||||
types = append(types, constants.TaskTypePollingRealname)
|
||||
}
|
||||
if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 {
|
||||
types = append(types, constants.TaskTypePollingCarddata)
|
||||
}
|
||||
if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 {
|
||||
types = append(types, constants.TaskTypePollingPackage)
|
||||
}
|
||||
if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 {
|
||||
types = append(types, constants.TaskTypePollingProtect)
|
||||
}
|
||||
if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 {
|
||||
types = append(types, constants.TaskTypePollingCardStatus)
|
||||
}
|
||||
return types
|
||||
}
|
||||
|
||||
// calcInitialDelay 计算新入队卡的初始检查时间(加随机抖动避免惊群)
|
||||
// 抖动范围 [0, max(interval/10, 2)) 秒;下界取 2 保证 rand.Intn 有实际随机性
|
||||
// (rand.Intn(1) 永远返回 0,对 interval≤10 的配置会导致零抖动)
|
||||
func calcInitialDelay(_ *model.IotCard, cfg *model.PollingConfig, taskType string) time.Time {
|
||||
func calcInitialDelay(interval int) time.Time {
|
||||
now := time.Now()
|
||||
var interval int
|
||||
switch taskType {
|
||||
case constants.TaskTypePollingRealname:
|
||||
if cfg.RealnameCheckInterval != nil {
|
||||
interval = *cfg.RealnameCheckInterval
|
||||
}
|
||||
case constants.TaskTypePollingCarddata:
|
||||
if cfg.CarddataCheckInterval != nil {
|
||||
interval = *cfg.CarddataCheckInterval
|
||||
}
|
||||
case constants.TaskTypePollingPackage:
|
||||
if cfg.PackageCheckInterval != nil {
|
||||
interval = *cfg.PackageCheckInterval
|
||||
}
|
||||
case constants.TaskTypePollingProtect:
|
||||
if cfg.ProtectCheckInterval != nil {
|
||||
interval = *cfg.ProtectCheckInterval
|
||||
}
|
||||
case constants.TaskTypePollingCardStatus:
|
||||
if cfg.CardStatusCheckInterval != nil {
|
||||
interval = *cfg.CardStatusCheckInterval
|
||||
}
|
||||
}
|
||||
if interval <= 0 {
|
||||
return now
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ type Scheduler struct {
|
||||
|
||||
packageActivationHandler *PackageActivationHandler
|
||||
dataResetHandler *DataResetHandler
|
||||
initializer *PollingInitializer // 可选,nil 表示不守卫 Init 完成
|
||||
|
||||
stopChan chan struct{}
|
||||
wg sync.WaitGroup
|
||||
@@ -103,9 +104,29 @@ func (s *Scheduler) SetStopResumeCallback(callback packagepkg.StopResumeCallback
|
||||
}
|
||||
}
|
||||
|
||||
// SetInitializer 注入初始化器(可选,在 Start 前调用)
|
||||
// Init 未完成时调度器跳过分片出队,消除启动期无效轮询噪音
|
||||
func (s *Scheduler) SetInitializer(init *PollingInitializer) {
|
||||
s.initializer = init
|
||||
}
|
||||
|
||||
// writeHeartbeat 写入调度器心跳,表明调度器存活
|
||||
func (s *Scheduler) writeHeartbeat(ctx context.Context) {
|
||||
if err := s.redis.Set(ctx, constants.RedisPollingSchedulerHeartbeatKey(),
|
||||
time.Now().Unix(), 2*s.cfg.ScheduleInterval).Err(); err != nil {
|
||||
s.logger.Warn("写入调度器心跳失败", zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// scheduleLoop 调度循环
|
||||
func (s *Scheduler) scheduleLoop(ctx context.Context) {
|
||||
defer s.wg.Done()
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
s.logger.Error("调度主循环发生 panic,调度已停止,需重启 Worker",
|
||||
zap.Any("panic", r), zap.Stack("stack"))
|
||||
}
|
||||
}()
|
||||
|
||||
ticker := time.NewTicker(s.cfg.ScheduleInterval)
|
||||
activationTicker := time.NewTicker(10 * time.Second)
|
||||
@@ -123,6 +144,7 @@ func (s *Scheduler) scheduleLoop(ctx context.Context) {
|
||||
s.logger.Info("调度循环收到 ctx 取消信号")
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.writeHeartbeat(ctx)
|
||||
s.processShardSchedule(ctx)
|
||||
case <-activationTicker.C:
|
||||
s.processActivationTasks(ctx)
|
||||
@@ -133,10 +155,16 @@ func (s *Scheduler) scheduleLoop(ctx context.Context) {
|
||||
// processShardSchedule 处理手动队列和分片定时队列(每 1 秒触发)
|
||||
// 使用 90% 的 tick 间隔作为超时,确保单分片 Redis 挂起时不阻塞下一个 tick
|
||||
func (s *Scheduler) processShardSchedule(ctx context.Context) {
|
||||
// 手动队列不受 Init 影响(ConfigManager 已就绪即可)
|
||||
for _, taskType := range allTaskTypes {
|
||||
s.processManualQueue(ctx, taskType, s.cfg.MaxManualBatchSize)
|
||||
}
|
||||
|
||||
// Init 未完成时跳过分片扫描,避免空轮询噪音
|
||||
if s.initializer != nil && !s.initializer.IsCompleted() {
|
||||
return
|
||||
}
|
||||
|
||||
if s.queueMgr == nil {
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user