diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 1592d2d..741f057 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -314,10 +314,13 @@ func startPollingInitializer(ctx context.Context, runtime *workerRuntime, appLog pollingInitializer.StartBackground(ctx) runtime.pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) { - if !hadConfigs && hasConfigs { - appLogger.Info("轮询配置从空变为非空,触发队列重新初始化") + if hasConfigs { + appLogger.Info("轮询配置已变更,触发队列重新初始化", + zap.Bool("had_configs", hadConfigs)) pollingInitializer.Restart(ctx) + return } + appLogger.Info("轮询配置已清空,跳过队列重新初始化") }) return pollingInitializer diff --git a/internal/polling/initializer.go b/internal/polling/initializer.go index 61bcdd6..54a3b12 100644 --- a/internal/polling/initializer.go +++ b/internal/polling/initializer.go @@ -52,6 +52,7 @@ type PollingInitializer struct { progress initProgress initCompleted atomic.Bool + restartQueued atomic.Bool stopChan chan struct{} wg sync.WaitGroup @@ -94,11 +95,12 @@ func (p *PollingInitializer) IsCompleted() bool { return p.initCompleted.Load() } -// Restart 重新执行初始化(当轮询配置从空变为非空时调用) -// 使用 CAS 确保只有初始化已完成时才能重启,避免并发重入 +// Restart 重新执行初始化,用于配置变更后补齐新增任务类型的分片队列。 +// 若初始化正在进行中,则记录一次待重启请求,当前轮完成后再自动执行。 func (p *PollingInitializer) Restart(ctx context.Context) { if !p.initCompleted.CompareAndSwap(true, false) { - p.logger.Info("轮询初始化仍在进行中,跳过重启") + p.restartQueued.Store(true) + p.logger.Info("轮询初始化仍在进行中,已标记完成后重新初始化") return } p.setStatus("pending", "") @@ -200,6 +202,20 @@ func (p *PollingInitializer) run(ctx context.Context) { p.logger.Info("分片渐进式初始化完成", zap.Int64("total_loaded", snapshot.LoadedCards), zap.Duration("duration", time.Since(snapshot.StartTime))) + + if p.restartQueued.Swap(false) { + select { + case <-ctx.Done(): + return + default: + } + if p.initCompleted.CompareAndSwap(true, false) { + p.setStatus("pending", "") + p.wg.Add(1) + go p.run(ctx) + p.logger.Info("检测到初始化期间配置变更,完成后再次初始化") + } + } } // initBatch 使用 Pipeline 将一批卡写入分片队列和缓存 diff --git a/internal/service/polling/config_service.go b/internal/service/polling/config_service.go index 3a68a8c..a9d03f3 100644 --- a/internal/service/polling/config_service.go +++ b/internal/service/polling/config_service.go @@ -155,6 +155,7 @@ func (s *ConfigService) Update(ctx context.Context, id uint, req *dto.UpdatePoll return nil, errors.Wrap(errors.CodeInternalError, err, "更新轮询配置失败") } + s.notifyConfigChanged(ctx, "updated") return s.toResponse(config), nil }