diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 7ddd189..beebc50 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -157,6 +157,14 @@ func main() { pollingInitializer := polling.NewPollingInitializer(pollingIotCardStore, redisClient, pollingConfigMgr, pollingQueueMgr, appLogger) pollingInitializer.StartBackground(ctx) + // 订阅配置变更事件：配置从空变为非空时重启 Initializer，将卡重新入队 + pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) { + if !hadConfigs && hasConfigs { + appLogger.Info("轮询配置从空变为非空，触发队列重新初始化") + pollingInitializer.Restart(ctx) + } + }) + // 创建生命周期服务（Worker 进程用，功能更完整） pollingDeviceSimBindingStore := postgres.NewDeviceSimBindingStore(db, redisClient) pollingDeviceStore := postgres.NewDeviceStore(db, redisClient) diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 7f2ee39..536b1df 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -206,7 +206,7 @@ func initServices(s *stores, deps *Dependencies) *services { Order: orderSvc.New(deps.DB, deps.Redis, s.Order, s.OrderItem, s.AgentWallet, s.AssetWallet, s.Payment, purchaseValidation, s.ShopPackageAllocation, s.ShopSeriesAllocation, s.IotCard, s.Device, s.PackageSeries, s.PackageUsage, s.Package, wechatConfig, deps.WechatPayment, paymentLoader, deps.QueueClient, deps.Logger, s.AssetIdentifier, s.PersonalCustomer, s.PersonalCustomerPhone), Exchange: exchangeSvc.New(deps.DB, s.ExchangeOrder, s.IotCard, s.Device, s.AssetWallet, s.AssetWalletTransaction, s.PackageUsage, s.PackageUsageDailyRecord, s.ResourceTag, s.PersonalCustomerDevice, deps.Logger), Recharge: rechargeSvc.New(deps.DB, s.AssetWallet, s.AssetWalletTransaction, s.IotCard, s.Device, s.ShopSeriesAllocation, s.PackageSeries, s.CommissionRecord, wechatConfig, paymentLoader, deps.Logger), - PollingConfig: pollingSvc.NewConfigService(s.PollingConfig), + PollingConfig: pollingSvc.NewConfigService(s.PollingConfig, deps.Redis, deps.Logger), PollingConcurrency: pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis), PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis), PollingAlert: pollingSvc.NewAlertService(s.PollingAlertRule, s.PollingAlertHistory, deps.Redis, deps.Logger), diff --git a/internal/polling/config_manager.go b/internal/polling/config_manager.go index 13c4130..267a819 100644 --- a/internal/polling/config_manager.go +++ b/internal/polling/config_manager.go @@ -63,6 +63,45 @@ func (m *PollingConfigManager) Load(ctx context.Context) error { return nil } +// WatchChanges 订阅 Redis 配置变更频道，收到通知后立即刷新内存缓存 +// onRefreshed 回调参数：(变更前是否有配置, 变更后是否有配置) +// 当从无配置变为有配置时，调用方可据此触发 Initializer 重启 +func (m *PollingConfigManager) WatchChanges(ctx context.Context, onRefreshed func(hadConfigs, hasConfigs bool)) { + go func() { + pubsub := m.redis.Subscribe(ctx, constants.RedisPollingConfigChangedChannel()) + defer pubsub.Close() + + ch := pubsub.Channel() + for { + select { + case <-ctx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + m.mu.RLock() + hadConfigs := len(m.configs) > 0 + m.mu.RUnlock() + + m.logger.Info("收到配置变更通知，立即刷新内存缓存", zap.String("event", msg.Payload)) + if err := m.Load(ctx); err != nil { + m.logger.Warn("配置变更后刷新缓存失败，等待下次定时刷新", zap.Error(err)) + continue + } + + m.mu.RLock() + hasConfigs := len(m.configs) > 0 + m.mu.RUnlock() + + if onRefreshed != nil { + onRefreshed(hadConfigs, hasConfigs) + } + } + } + }() +} + // Start 启动定时刷新（每 5 分钟自动 Load 一次） // 使用 sync.Once 确保只启动一个刷新 goroutine func (m *PollingConfigManager) Start(ctx context.Context) { diff --git a/internal/polling/initializer.go b/internal/polling/initializer.go index ac29ed9..61bcdd6 100644 --- a/internal/polling/initializer.go +++ b/internal/polling/initializer.go @@ -94,6 +94,19 @@ func (p *PollingInitializer) IsCompleted() bool { return p.initCompleted.Load() } +// Restart 重新执行初始化（当轮询配置从空变为非空时调用） +// 使用 CAS 确保只有初始化已完成时才能重启，避免并发重入 +func (p *PollingInitializer) Restart(ctx context.Context) { + if !p.initCompleted.CompareAndSwap(true, false) { + p.logger.Info("轮询初始化仍在进行中，跳过重启") + return + } + p.setStatus("pending", "") + p.wg.Add(1) + go p.run(ctx) + p.logger.Info("轮询初始化已重新启动（配置变更触发）") +} + // GetProgress 返回当前初始化进度快照（加锁读取，返回值拷贝） func (p *PollingInitializer) GetProgress() InitProgress { p.progress.mu.RLock() diff --git a/internal/service/polling/config_service.go b/internal/service/polling/config_service.go index d3f0848..3a68a8c 100644 --- a/internal/service/polling/config_service.go +++ b/internal/service/polling/config_service.go @@ -4,6 +4,8 @@ import ( "context" "time" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/model" @@ -18,11 +20,22 @@ import ( // ConfigService 轮询配置服务 type ConfigService struct { configStore *postgres.PollingConfigStore + redis *redis.Client + logger *zap.Logger } // NewConfigService 创建轮询配置服务实例 -func NewConfigService(configStore *postgres.PollingConfigStore) *ConfigService { - return &ConfigService{configStore: configStore} +func NewConfigService(configStore *postgres.PollingConfigStore, redisClient *redis.Client, logger *zap.Logger) *ConfigService { + return &ConfigService{configStore: configStore, redis: redisClient, logger: logger} +} + +func (s *ConfigService) notifyConfigChanged(ctx context.Context, event string) { + if s.redis == nil { + return + } + if err := s.redis.Publish(ctx, constants.RedisPollingConfigChangedChannel(), event).Err(); err != nil { + s.logger.Warn("通知轮询配置变更失败", zap.String("event", event), zap.Error(err)) + } } // Create 创建轮询配置 @@ -66,6 +79,7 @@ func (s *ConfigService) Create(ctx context.Context, req *dto.CreatePollingConfig return nil, errors.Wrap(errors.CodeInternalError, err, "创建轮询配置失败") } + s.notifyConfigChanged(ctx, "created") return s.toResponse(config), nil } @@ -158,6 +172,7 @@ func (s *ConfigService) Delete(ctx context.Context, id uint) error { return errors.Wrap(errors.CodeInternalError, err, "删除轮询配置失败") } + s.notifyConfigChanged(ctx, "deleted") return nil } @@ -224,6 +239,7 @@ func (s *ConfigService) UpdateStatus(ctx context.Context, id uint, status int16) return errors.Wrap(errors.CodeInternalError, err, "更新轮询配置状态失败") } + s.notifyConfigChanged(ctx, "updated") return nil } diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index a36c670..3b66920 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -435,6 +435,13 @@ func RedisPollingSchedulerHeartbeatKey() string { return "polling:scheduler:heartbeat" } +// RedisPollingConfigChangedChannel 轮询配置变更 Pub/Sub 频道 +// 用途：API 进程发布配置变更事件，Worker 进程订阅后立即刷新内存缓存 +// 消息内容：事件类型字符串（"created"/"updated"/"deleted"），仅用于日志，不影响逻辑 +func RedisPollingConfigChangedChannel() string { + return "polling:config:changed" +} + // ======================================== // 支付配置缓存 Redis Key // ========================================