// Package carrierqueue 实现按运营商账户调度的卡轮询结构:账户延时队列、卡到期信息、领取 / 续期 / 收尾 / 请求检查 Lua、 // 分发器、初始化与受控系统配置。Redis 是唯一权威状态,NSQ 只做传输(add-carrier-polling-scheduler)。 package carrierqueue import ( "context" "encoding/json" stderrors "errors" "strconv" "sync" "time" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/infrastructure/systemconfig" "github.com/break/junhong_cmp_fiber/pkg/constants" ) // settingsTTL 是调度参数在进程内的缓存时长;后台修改数秒内生效。 const settingsTTL = 5 * time.Second // RegisterConfigDefinitions 注册按运营商账户调度的受控系统配置定义(含 JSON 结构校验)。 func RegisterConfigDefinitions(registry *systemconfig.Registry, logger *zap.Logger) { if registry == nil { return } minOne, maxSeconds, maxCount := int64(1), int64(86400*7), int64(100000) definitions := []systemconfig.Definition{ {Key: constants.SystemConfigPollingSchedulerMode, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeString, DefaultValue: constants.PollingSchedulerModeLegacy, Description: "卡轮询调度模式:legacy 旧分片调度,carrier 按运营商账户调度(整体切换)", EnumValues: []string{constants.PollingSchedulerModeLegacy, constants.PollingSchedulerModeCarrier}}, {Key: constants.SystemConfigPollingStepOrder, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, DefaultValue: `["card_status","carddata","realname","package"]`, Description: "卡任务步骤顺序", Validator: validateStepOrder}, {Key: constants.SystemConfigPollingStepTimeout, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, DefaultValue: `{"card_status":15,"carddata":15,"realname":15,"package":20}`, Description: "各检查类型单步超时(秒)", Validator: validateStepTimeout}, {Key: constants.SystemConfigPollingTaskTimeoutMargin, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, DefaultValue: "10", Description: "卡任务总超时余量(秒)", Min: &minOne, Max: &maxSeconds}, {Key: constants.SystemConfigPollingSkipPackageAfterFlow, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeBool, DefaultValue: "true", Description: "同轮流量步骤已触发停复机评估时跳过套餐步骤"}, {Key: constants.SystemConfigPollingFailureBackoff, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, DefaultValue: `{"default":[60,300,900,1800,3600]}`, Description: "失败退避序列(秒,递增),可按检查类型覆盖 default", Validator: validateBackoff}, {Key: constants.SystemConfigPollingAlertFailCount, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, DefaultValue: `{"default":5}`, Description: "连续失败告警阈值,可按检查类型覆盖 default", Validator: validateAlertCount}, {Key: constants.SystemConfigPollingAlertFailDuration, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, DefaultValue: "21600", Description: "持续失败时长告警阈值(秒)", Min: &minOne, Max: &maxSeconds}, {Key: constants.SystemConfigPollingImmediateAlert, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, DefaultValue: `["account_config"]`, Description: "首次出现即告警的错误类别", Validator: validateCategories}, {Key: constants.SystemConfigPollingAccountAlertWindow, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, DefaultValue: "600", Description: "账户级告警统计窗口(秒)", Min: &minOne, Max: &maxSeconds}, {Key: constants.SystemConfigPollingAccountAlertThreshold, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, DefaultValue: "50", Description: "账户级告警阈值:窗口内同账户新增失败告警数", Min: &minOne, Max: &maxCount}, } for _, definition := range definitions { if _, exists := registry.Get(definition.Key); exists { continue } if err := registry.Register(definition); err != nil && logger != nil { logger.Error("注册按运营商账户调度系统配置失败,将使用代码默认值", zap.String("config_key", definition.Key), zap.Error(err)) } } } func validCheckType(value string) bool { switch value { case constants.CheckTypeCardStatus, constants.CheckTypeCarddata, constants.CheckTypeRealname, constants.CheckTypePackage: return true } return false } func validateStepOrder(raw string) error { var steps []string if err := json.Unmarshal([]byte(raw), &steps); err != nil { return stderrors.New("步骤顺序必须是字符串数组") } if len(steps) == 0 { return stderrors.New("步骤顺序不能为空") } seen := map[string]bool{} for _, step := range steps { if !validCheckType(step) { return stderrors.New("步骤只能是 card_status、carddata、realname、package") } if seen[step] { return stderrors.New("步骤不能重复") } seen[step] = true } return nil } func validateStepTimeout(raw string) error { var timeouts map[string]int if err := json.Unmarshal([]byte(raw), &timeouts); err != nil { return stderrors.New("单步超时必须是 {检查类型: 秒} 对象") } for key, value := range timeouts { if !validCheckType(key) { return stderrors.New("单步超时的检查类型非法") } if value <= 0 || value > 600 { return stderrors.New("单步超时必须在 1~600 秒之间") } } return nil } func validateBackoff(raw string) error { var backoff map[string][]int if err := json.Unmarshal([]byte(raw), &backoff); err != nil { return stderrors.New("退避序列必须是 {default 或检查类型: [秒...]} 对象") } if len(backoff["default"]) == 0 { return stderrors.New("退避序列必须包含非空的 default") } for key, values := range backoff { if key != "default" && !validCheckType(key) { return stderrors.New("退避序列的检查类型非法") } for index, value := range values { if value <= 0 { return stderrors.New("退避间隔必须为正数") } if index > 0 && value <= values[index-1] { return stderrors.New("退避序列必须递增") } } } return nil } func validateAlertCount(raw string) error { var counts map[string]int if err := json.Unmarshal([]byte(raw), &counts); err != nil { return stderrors.New("告警阈值必须是 {default 或检查类型: 次数} 对象") } if counts["default"] <= 0 { return stderrors.New("告警阈值必须包含正数 default") } for key, value := range counts { if key != "default" && !validCheckType(key) { return stderrors.New("告警阈值的检查类型非法") } if value <= 0 { return stderrors.New("告警阈值必须为正数") } } return nil } func validateCategories(raw string) error { var categories []string if err := json.Unmarshal([]byte(raw), &categories); err != nil { return stderrors.New("错误类别必须是字符串数组") } return nil } // Settings 是按运营商账户调度的运行参数快照。 type Settings struct { Mode string StepOrder []string StepTimeouts map[string]time.Duration TaskTimeoutMargin time.Duration SkipPackageAfterFlow bool Backoff map[string][]time.Duration AlertFailCount map[string]int AlertFailDuration time.Duration ImmediateAlert map[string]bool AccountAlertWindow time.Duration AccountAlertLimit int } // CarrierMode 判断当前是否为按运营商账户调度。 func (s Settings) CarrierMode() bool { return s.Mode == constants.PollingSchedulerModeCarrier } // StepTimeout 返回检查类型单步超时;未配置时 15 秒。 func (s Settings) StepTimeout(checkType string) time.Duration { if value, ok := s.StepTimeouts[checkType]; ok && value > 0 { return value } return 15 * time.Second } // TaskTimeout 返回卡任务总超时 = 各步骤超时之和 + 余量。 func (s Settings) TaskTimeout() time.Duration { total := s.TaskTimeoutMargin for _, step := range s.StepOrder { total += s.StepTimeout(step) } return total } // BackoffFor 返回检查类型第 n 次连续失败(n 从 1 起)的退避间隔,不超过 maxInterval(> 0 时)。 func (s Settings) BackoffFor(checkType string, failCount int, maxInterval time.Duration) time.Duration { sequence := s.Backoff[checkType] if len(sequence) == 0 { sequence = s.Backoff["default"] } if len(sequence) == 0 { sequence = []time.Duration{time.Minute} } index := failCount - 1 if index < 0 { index = 0 } if index >= len(sequence) { index = len(sequence) - 1 } delay := sequence[index] if maxInterval > 0 && delay > maxInterval { delay = maxInterval } return delay } // AlertThreshold 返回检查类型连续失败告警阈值。 func (s Settings) AlertThreshold(checkType string) int { if value, ok := s.AlertFailCount[checkType]; ok && value > 0 { return value } if value := s.AlertFailCount["default"]; value > 0 { return value } return 5 } // DefaultSettings 是配置读取失败时的安全默认值(旧分片调度)。 func DefaultSettings() Settings { return Settings{ Mode: constants.PollingSchedulerModeLegacy, StepOrder: constants.AllCheckTypes(), StepTimeouts: map[string]time.Duration{ constants.CheckTypeCardStatus: 15 * time.Second, constants.CheckTypeCarddata: 15 * time.Second, constants.CheckTypeRealname: 15 * time.Second, constants.CheckTypePackage: 20 * time.Second, }, TaskTimeoutMargin: 10 * time.Second, SkipPackageAfterFlow: true, Backoff: map[string][]time.Duration{"default": {time.Minute, 5 * time.Minute, 15 * time.Minute, 30 * time.Minute, time.Hour}}, AlertFailCount: map[string]int{"default": 5}, AlertFailDuration: 6 * time.Hour, ImmediateAlert: map[string]bool{constants.GatewayClassAccountConfig: true}, AccountAlertWindow: 10 * time.Minute, AccountAlertLimit: 50, } } // ConfigReader 读取已注册的受控系统配置。 type ConfigReader interface { Get(ctx context.Context, key string) (string, error) } // SettingsProvider 在进程内缓存调度参数,超过 TTL 重新读取。 type SettingsProvider struct { reader ConfigReader logger *zap.Logger mu sync.Mutex cached Settings loaded time.Time } // NewSettingsProvider 创建调度参数提供者;reader 为 nil 时始终返回默认值。 func NewSettingsProvider(reader ConfigReader, logger *zap.Logger) *SettingsProvider { if logger == nil { logger = zap.NewNop() } return &SettingsProvider{reader: reader, logger: logger, cached: DefaultSettings()} } // Current 返回当前调度参数;读取失败时保留上一次成功值。 func (p *SettingsProvider) Current(ctx context.Context) Settings { if p == nil { return DefaultSettings() } p.mu.Lock() defer p.mu.Unlock() if p.reader == nil || time.Since(p.loaded) < settingsTTL { return p.cached } next, err := p.load(ctx) p.loaded = time.Now() if err != nil { p.logger.Warn("读取按运营商账户调度配置失败,沿用上一次配置", zap.Error(err)) return p.cached } p.cached = next return next } // Invalidate 使缓存立即失效(后台保存后调用)。 func (p *SettingsProvider) Invalidate() { if p == nil { return } p.mu.Lock() p.loaded = time.Time{} p.mu.Unlock() } func (p *SettingsProvider) load(ctx context.Context) (Settings, error) { s := DefaultSettings() get := func(key string) (string, error) { return p.reader.Get(ctx, key) } raw, err := get(constants.SystemConfigPollingSchedulerMode) if err != nil { return s, err } s.Mode = raw if raw, err = get(constants.SystemConfigPollingStepOrder); err != nil { return s, err } if err := json.Unmarshal([]byte(raw), &s.StepOrder); err != nil || validateStepOrder(raw) != nil { return s, stderrors.New("步骤顺序配置非法") } if raw, err = get(constants.SystemConfigPollingStepTimeout); err != nil { return s, err } var timeouts map[string]int if err := json.Unmarshal([]byte(raw), &timeouts); err != nil { return s, stderrors.New("单步超时配置非法") } for key, value := range timeouts { s.StepTimeouts[key] = time.Duration(value) * time.Second } if raw, err = get(constants.SystemConfigPollingFailureBackoff); err != nil { return s, err } var backoff map[string][]int if err := json.Unmarshal([]byte(raw), &backoff); err != nil { return s, stderrors.New("退避序列配置非法") } s.Backoff = map[string][]time.Duration{} for key, values := range backoff { for _, value := range values { s.Backoff[key] = append(s.Backoff[key], time.Duration(value)*time.Second) } } if raw, err = get(constants.SystemConfigPollingAlertFailCount); err != nil { return s, err } if err := json.Unmarshal([]byte(raw), &s.AlertFailCount); err != nil { return s, stderrors.New("告警阈值配置非法") } if raw, err = get(constants.SystemConfigPollingImmediateAlert); err != nil { return s, err } var categories []string if err := json.Unmarshal([]byte(raw), &categories); err != nil { return s, stderrors.New("首次告警类别配置非法") } s.ImmediateAlert = map[string]bool{} for _, category := range categories { s.ImmediateAlert[category] = true } if raw, err = get(constants.SystemConfigPollingSkipPackageAfterFlow); err != nil { return s, err } s.SkipPackageAfterFlow = raw == "true" ints := []struct { key string set func(int) }{ {constants.SystemConfigPollingTaskTimeoutMargin, func(v int) { s.TaskTimeoutMargin = time.Duration(v) * time.Second }}, {constants.SystemConfigPollingAlertFailDuration, func(v int) { s.AlertFailDuration = time.Duration(v) * time.Second }}, {constants.SystemConfigPollingAccountAlertWindow, func(v int) { s.AccountAlertWindow = time.Duration(v) * time.Second }}, {constants.SystemConfigPollingAccountAlertThreshold, func(v int) { s.AccountAlertLimit = v }}, } for _, item := range ints { raw, err := get(item.key) if err != nil { return s, err } value, convErr := strconv.Atoi(raw) if convErr != nil { return s, stderrors.New("调度整数配置非法") } item.set(value) } return s, nil }