2fe7c1f8ac
- 四个轮询处理器拆为步骤函数 Step（返回结果分类），旧 Asynq Handle 只做并发控制与原有重入队映射，行为不变；carrier 模式下旧 handler 直接结束
- 网关限流增加等待模式（ctx 标记），NSQ 消费者等名额到单步超时，超时按限流处理不计失败；记录限流等待次数
- 卡任务执行器：核对 token 并抢执行权，按配置步骤顺序串行执行到期检查（单步超时、总超时、panic 恢复、流量已触发评估时跳过套餐），执行中续期并 Touch，续期失败立即停止；收尾写回到期、连续失败、退避、延迟确认与告警
- 分发器：每秒按账户抢分发锁领取（执行上限按消费能力收紧、优先占比、按限流组剩余 QPS 估算），发布到 polling.card.{账户}，发布失败释放；消费旧优先提示与手动触发；心跳与处理中标记到期计数
- NSQ 生产者与按账户动态订阅的消费者管理器，优雅退出宽限期
- 失败告警异步写 tb_polling_failure_alert（同卡同类型一条未处理，成功转已恢复），账户窗口内告警数超阈值产生账户级告警
- 模式监听：切 carrier 时旧分片调度停止出队，加锁初始化新结构后开始领取并订阅；回 legacy 时停止消费、重建旧分片队列；生命周期（API 与 Worker）在 carrier 模式写账户队列
393 lines
15 KiB
Go
393 lines
15 KiB
Go
// Package carrierqueue 实现按运营商账户调度的卡轮询结构：账户延时队列、卡到期信息、领取 / 续期 / 收尾 / 请求检查 Lua、
|
||
// 分发器、初始化与受控系统配置。Redis 是唯一权威状态，NSQ 只做传输（add-carrier-polling-scheduler）。
|
||
package carrierqueue
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
stderrors "errors"
|
||
"strconv"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"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
|
||
}
|
||
|
||
// NewDefaultSettingsProvider 用独立注册表与读取器创建调度参数提供者（API 与 Worker 共用同一份数据库与缓存数据）。
|
||
func NewDefaultSettingsProvider(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *SettingsProvider {
|
||
registry := systemconfig.NewRegistry()
|
||
RegisterConfigDefinitions(registry, logger)
|
||
var cache systemconfig.Cache
|
||
if redisClient != nil {
|
||
cache = systemconfig.NewRedisCache(redisClient)
|
||
}
|
||
reader := systemconfig.NewReader(db, registry, cache, systemconfig.NewLogAlertSink(logger))
|
||
return NewSettingsProvider(reader, logger)
|
||
}
|