fix: 修复禁用所有轮询配置重启后再启用无法工作的问题
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m23s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m23s
- 新增 RedisPollingConfigChangedChannel 常量,用于跨进程配置变更通知 - ConfigService 注入 Redis,Create/Delete/UpdateStatus 后发布 Pub/Sub 事件 - PollingConfigManager 新增 WatchChanges() 方法,收到通知立即刷新内存缓存 - PollingInitializer 新增 Restart() 方法,支持初始化完成后安全重启(CAS 防并发) - Worker 订阅配置变更事件,configs 从空变为非空时触发 Initializer.Restart() 根因:启动时所有配置禁用导致分片队列为空,Initializer 是一次性的, 之后启用配置只更新 DB 但不重建队列,调度器无卡可处理
This commit is contained in:
@@ -157,6 +157,14 @@ func main() {
|
|||||||
pollingInitializer := polling.NewPollingInitializer(pollingIotCardStore, redisClient, pollingConfigMgr, pollingQueueMgr, appLogger)
|
pollingInitializer := polling.NewPollingInitializer(pollingIotCardStore, redisClient, pollingConfigMgr, pollingQueueMgr, appLogger)
|
||||||
pollingInitializer.StartBackground(ctx)
|
pollingInitializer.StartBackground(ctx)
|
||||||
|
|
||||||
|
// 订阅配置变更事件:配置从空变为非空时重启 Initializer,将卡重新入队
|
||||||
|
pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) {
|
||||||
|
if !hadConfigs && hasConfigs {
|
||||||
|
appLogger.Info("轮询配置从空变为非空,触发队列重新初始化")
|
||||||
|
pollingInitializer.Restart(ctx)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
// 创建生命周期服务(Worker 进程用,功能更完整)
|
// 创建生命周期服务(Worker 进程用,功能更完整)
|
||||||
pollingDeviceSimBindingStore := postgres.NewDeviceSimBindingStore(db, redisClient)
|
pollingDeviceSimBindingStore := postgres.NewDeviceSimBindingStore(db, redisClient)
|
||||||
pollingDeviceStore := postgres.NewDeviceStore(db, redisClient)
|
pollingDeviceStore := postgres.NewDeviceStore(db, redisClient)
|
||||||
|
|||||||
@@ -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),
|
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),
|
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),
|
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),
|
PollingConcurrency: pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis),
|
||||||
PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis),
|
PollingMonitoring: pollingSvc.NewMonitoringService(deps.Redis),
|
||||||
PollingAlert: pollingSvc.NewAlertService(s.PollingAlertRule, s.PollingAlertHistory, deps.Redis, deps.Logger),
|
PollingAlert: pollingSvc.NewAlertService(s.PollingAlertRule, s.PollingAlertHistory, deps.Redis, deps.Logger),
|
||||||
|
|||||||
@@ -63,6 +63,45 @@ func (m *PollingConfigManager) Load(ctx context.Context) error {
|
|||||||
return nil
|
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 一次)
|
// Start 启动定时刷新(每 5 分钟自动 Load 一次)
|
||||||
// 使用 sync.Once 确保只启动一个刷新 goroutine
|
// 使用 sync.Once 确保只启动一个刷新 goroutine
|
||||||
func (m *PollingConfigManager) Start(ctx context.Context) {
|
func (m *PollingConfigManager) Start(ctx context.Context) {
|
||||||
|
|||||||
@@ -94,6 +94,19 @@ func (p *PollingInitializer) IsCompleted() bool {
|
|||||||
return p.initCompleted.Load()
|
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 返回当前初始化进度快照(加锁读取,返回值拷贝)
|
// GetProgress 返回当前初始化进度快照(加锁读取,返回值拷贝)
|
||||||
func (p *PollingInitializer) GetProgress() InitProgress {
|
func (p *PollingInitializer) GetProgress() InitProgress {
|
||||||
p.progress.mu.RLock()
|
p.progress.mu.RLock()
|
||||||
|
|||||||
@@ -4,6 +4,8 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/redis/go-redis/v9"
|
||||||
|
"go.uber.org/zap"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
|
|
||||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||||
@@ -18,11 +20,22 @@ import (
|
|||||||
// ConfigService 轮询配置服务
|
// ConfigService 轮询配置服务
|
||||||
type ConfigService struct {
|
type ConfigService struct {
|
||||||
configStore *postgres.PollingConfigStore
|
configStore *postgres.PollingConfigStore
|
||||||
|
redis *redis.Client
|
||||||
|
logger *zap.Logger
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewConfigService 创建轮询配置服务实例
|
// NewConfigService 创建轮询配置服务实例
|
||||||
func NewConfigService(configStore *postgres.PollingConfigStore) *ConfigService {
|
func NewConfigService(configStore *postgres.PollingConfigStore, redisClient *redis.Client, logger *zap.Logger) *ConfigService {
|
||||||
return &ConfigService{configStore: configStore}
|
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 创建轮询配置
|
// Create 创建轮询配置
|
||||||
@@ -66,6 +79,7 @@ func (s *ConfigService) Create(ctx context.Context, req *dto.CreatePollingConfig
|
|||||||
return nil, errors.Wrap(errors.CodeInternalError, err, "创建轮询配置失败")
|
return nil, errors.Wrap(errors.CodeInternalError, err, "创建轮询配置失败")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
s.notifyConfigChanged(ctx, "created")
|
||||||
return s.toResponse(config), nil
|
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, "删除轮询配置失败")
|
return errors.Wrap(errors.CodeInternalError, err, "删除轮询配置失败")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
s.notifyConfigChanged(ctx, "deleted")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -224,6 +239,7 @@ func (s *ConfigService) UpdateStatus(ctx context.Context, id uint, status int16)
|
|||||||
return errors.Wrap(errors.CodeInternalError, err, "更新轮询配置状态失败")
|
return errors.Wrap(errors.CodeInternalError, err, "更新轮询配置状态失败")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
s.notifyConfigChanged(ctx, "updated")
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -435,6 +435,13 @@ func RedisPollingSchedulerHeartbeatKey() string {
|
|||||||
return "polling:scheduler:heartbeat"
|
return "polling:scheduler:heartbeat"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RedisPollingConfigChangedChannel 轮询配置变更 Pub/Sub 频道
|
||||||
|
// 用途:API 进程发布配置变更事件,Worker 进程订阅后立即刷新内存缓存
|
||||||
|
// 消息内容:事件类型字符串("created"/"updated"/"deleted"),仅用于日志,不影响逻辑
|
||||||
|
func RedisPollingConfigChangedChannel() string {
|
||||||
|
return "polling:config:changed"
|
||||||
|
}
|
||||||
|
|
||||||
// ========================================
|
// ========================================
|
||||||
// 支付配置缓存 Redis Key
|
// 支付配置缓存 Redis Key
|
||||||
// ========================================
|
// ========================================
|
||||||
|
|||||||
Reference in New Issue
Block a user