feat: 按运营商账户调度的分发器、卡任务执行器与模式切换
- 四个轮询处理器拆为步骤函数 Step（返回结果分类），旧 Asynq Handle 只做并发控制与原有重入队映射，行为不变；carrier 模式下旧 handler 直接结束
- 网关限流增加等待模式（ctx 标记），NSQ 消费者等名额到单步超时，超时按限流处理不计失败；记录限流等待次数
- 卡任务执行器：核对 token 并抢执行权，按配置步骤顺序串行执行到期检查（单步超时、总超时、panic 恢复、流量已触发评估时跳过套餐），执行中续期并 Touch，续期失败立即停止；收尾写回到期、连续失败、退避、延迟确认与告警
- 分发器：每秒按账户抢分发锁领取（执行上限按消费能力收紧、优先占比、按限流组剩余 QPS 估算），发布到 polling.card.{账户}，发布失败释放；消费旧优先提示与手动触发；心跳与处理中标记到期计数
- NSQ 生产者与按账户动态订阅的消费者管理器，优雅退出宽限期
- 失败告警异步写 tb_polling_failure_alert（同卡同类型一条未处理，成功转已恢复），账户窗口内告警数超阈值产生账户级告警
- 模式监听：切 carrier 时旧分片调度停止出队，加锁初始化新结构后开始领取并订阅；回 legacy 时停止消费、重建旧分片队列；生命周期（API 与 Worker）在 carrier 模式写账户队列
This commit is contained in:
邹进
228
cmd/worker/carrier_scheduler.go
Normal file
228
cmd/worker/carrier_scheduler.go
Normal file
@@ -0,0 +1,228 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/nsqbus"
|
||||
"github.com/break/junhong_cmp_fiber/internal/polling"
|
||||
"github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue"
|
||||
"github.com/break/junhong_cmp_fiber/internal/task"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/config"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/queue"
|
||||
)
|
||||
|
||||
// carrierInitLockKey 新结构初始化锁：多实例时只由一个实例初始化。
|
||||
const carrierInitLockKey = "polling:carrier:init_lock"
|
||||
|
||||
// carrierSchedulerRuntime 是按运营商账户调度在 Worker 内的运行组件：模式监听、初始化、分发器与 NSQ 消费者。
|
||||
type carrierSchedulerRuntime struct {
|
||||
cfg config.CarrierSchedulerConfig
|
||||
runtime *workerRuntime
|
||||
initializer *polling.CarrierInitializer
|
||||
dispatcher *carrierqueue.Dispatcher
|
||||
consumers *nsqbus.ConsumerManager
|
||||
producer *nsqbus.Producer
|
||||
alerts *carrierqueue.AlertWriter
|
||||
executor *task.CarrierCardExecutor
|
||||
singleton bool
|
||||
instanceID string
|
||||
logger *zap.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
lastMode string
|
||||
reinitQueued bool
|
||||
legacyRestarts func()
|
||||
}
|
||||
|
||||
// newCarrierSchedulerParts 在 Worker 运行时初始化阶段创建调度参数、账户队列存储与模式接缝，
|
||||
// 并注入旧调度相关组件（生命周期、轮询基类），使其在启动前即可按模式写入正确结构。
|
||||
func newCarrierSchedulerParts(runtime *workerRuntime, appLogger *zap.Logger) {
|
||||
settings := carrierqueue.NewDefaultSettingsProvider(runtime.db, runtime.redisClient, appLogger)
|
||||
store := carrierqueue.NewStore(runtime.redisClient)
|
||||
scheduling := polling.NewCarrierScheduling(store, settings, runtime.pollingConfigMgr, appLogger)
|
||||
runtime.carrierSettings = settings
|
||||
runtime.carrierStore = store
|
||||
runtime.carrierScheduling = scheduling
|
||||
runtime.lifecycleSvc.SetCarrierScheduling(scheduling)
|
||||
runtime.pollingBase.SetSchedulerModeReader(func(ctx context.Context) string { return settings.Current(ctx).Mode })
|
||||
if runtime.gatewayLimit != nil && runtime.gatewayLimit.Classifier != nil {
|
||||
classifier := runtime.gatewayLimit.Classifier
|
||||
runtime.pollingBase.SetErrorClassifier(func(ctx context.Context, op, message string) string {
|
||||
return classifier.Classify(ctx, op, message).Category
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// startCarrierScheduler 启动模式监听；分发器只在 leader/all 运行（多实例安全），消费者在所有处理任务的实例运行。
|
||||
func startCarrierScheduler(ctx context.Context, cfg *config.Config, runtime *workerRuntime, steps queue.PollingStepHandlers,
|
||||
singleton bool, legacyRestart func(), appLogger *zap.Logger) *carrierSchedulerRuntime {
|
||||
carrierCfg := cfg.Polling.Carrier
|
||||
instanceID := cfg.Worker.InstanceName
|
||||
if instanceID == "" {
|
||||
instanceID, _ = os.Hostname()
|
||||
}
|
||||
instanceID += "-" + strconv.Itoa(os.Getpid())
|
||||
alerts := carrierqueue.NewAlertWriter(runtime.db, runtime.redisClient, runtime.carrierSettings, appLogger)
|
||||
alerts.Start(ctx)
|
||||
executor := task.NewCarrierCardExecutor(runtime.pollingBase, runtime.carrierStore, runtime.carrierSettings, alerts,
|
||||
steps.Realname, steps.Carddata, steps.CardStatus, steps.Package, carrierCfg, instanceID)
|
||||
c := &carrierSchedulerRuntime{
|
||||
cfg: carrierCfg, runtime: runtime, singleton: singleton, instanceID: instanceID, logger: appLogger,
|
||||
alerts: alerts, executor: executor, legacyRestarts: legacyRestart,
|
||||
initializer: polling.NewCarrierInitializer(runtime.carrierScheduling, runtime.pollingIotCardStore, appLogger),
|
||||
}
|
||||
if carrierCfg.NSQDAddress != "" {
|
||||
producer, err := nsqbus.NewProducer(carrierCfg.NSQDAddress, appLogger)
|
||||
if err != nil {
|
||||
appLogger.Error("创建 NSQ 生产者失败，按运营商账户调度不可用", zap.Error(err))
|
||||
} else {
|
||||
c.producer = producer
|
||||
}
|
||||
c.consumers = nsqbus.NewConsumerManager(nsqbus.ConsumerConfig{
|
||||
NSQDAddress: carrierCfg.NSQDAddress, NSQLookupdAddresses: nsqbus.SplitAddresses(carrierCfg.NSQLookupdAddresses),
|
||||
MsgTimeout: carrierCfg.MsgTimeout, MaxAttempts: carrierCfg.MaxAttempts,
|
||||
}, constants.NSQChannelWorker, executor.Handle, appLogger)
|
||||
}
|
||||
var budgets carrierqueue.BudgetProvider
|
||||
if runtime.gatewayLimit != nil && runtime.gatewayLimit.Limiter != nil {
|
||||
budgets = runtime.gatewayLimit.Limiter
|
||||
}
|
||||
if c.producer != nil {
|
||||
c.dispatcher = carrierqueue.NewDispatcher(runtime.db, runtime.carrierStore, runtime.carrierSettings, c.producer, budgets,
|
||||
carrierCfg, instanceID, appLogger)
|
||||
if singleton {
|
||||
go c.dispatcher.Run(ctx)
|
||||
}
|
||||
}
|
||||
go c.watch(ctx)
|
||||
return c
|
||||
}
|
||||
|
||||
// RequestReinit 轮询配置变更时请求按新配置重新初始化新结构（仅 carrier 模式生效）。
|
||||
func (c *carrierSchedulerRuntime) RequestReinit() {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.reinitQueued = true
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
func (c *carrierSchedulerRuntime) watch(ctx context.Context) {
|
||||
ticker := time.NewTicker(3 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
c.step(ctx)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
c.shutdown()
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *carrierSchedulerRuntime) step(ctx context.Context) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
c.logger.Error("按运营商账户调度模式监听 panic，已恢复", zap.Any("panic", r))
|
||||
}
|
||||
}()
|
||||
mode := c.runtime.carrierSettings.Current(ctx).Mode
|
||||
c.mu.Lock()
|
||||
previous := c.lastMode
|
||||
c.lastMode = mode
|
||||
reinit := c.reinitQueued
|
||||
c.reinitQueued = false
|
||||
c.mu.Unlock()
|
||||
if mode == constants.PollingSchedulerModeCarrier {
|
||||
if previous != mode {
|
||||
c.logger.Info("卡轮询调度模式：按运营商账户调度", zap.String("previous", previous))
|
||||
}
|
||||
if c.singleton {
|
||||
c.ensureInitialized(ctx, reinit)
|
||||
}
|
||||
c.syncConsumers(ctx)
|
||||
return
|
||||
}
|
||||
if previous == constants.PollingSchedulerModeCarrier {
|
||||
c.logger.Info("卡轮询调度模式回到旧分片调度：停止领取与消费，重建旧分片队列")
|
||||
if c.consumers != nil {
|
||||
c.consumers.StopAll(c.grace())
|
||||
c.consumers.Reset()
|
||||
}
|
||||
if c.singleton {
|
||||
c.runtime.redisClient.Del(ctx, constants.RedisCarrierSchedulerInitStateKey)
|
||||
if c.legacyRestarts != nil {
|
||||
c.legacyRestarts()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ensureInitialized 切到 carrier 后初始化新结构；初始化完成前分发器不领取。多实例时用初始化锁协调。
|
||||
func (c *carrierSchedulerRuntime) ensureInitialized(ctx context.Context, reinit bool) {
|
||||
redisClient := c.runtime.redisClient
|
||||
state, _ := redisClient.Get(ctx, constants.RedisCarrierSchedulerInitStateKey).Result()
|
||||
if state == carrierqueue.InitStateDone && !reinit {
|
||||
return
|
||||
}
|
||||
ok, err := redisClient.SetNX(ctx, carrierInitLockKey, c.instanceID, 30*time.Minute).Result()
|
||||
if err != nil || !ok {
|
||||
return
|
||||
}
|
||||
defer redisClient.Del(context.WithoutCancel(ctx), carrierInitLockKey)
|
||||
if state != carrierqueue.InitStateDone {
|
||||
redisClient.Set(ctx, constants.RedisCarrierSchedulerInitStateKey, carrierqueue.InitStateRunning, 0)
|
||||
}
|
||||
started := time.Now()
|
||||
written, initErr := c.initializer.Run(ctx)
|
||||
if initErr != nil {
|
||||
c.logger.Error("按运营商账户调度初始化失败，下一轮重试", zap.Error(initErr))
|
||||
return
|
||||
}
|
||||
redisClient.Set(ctx, constants.RedisCarrierSchedulerInitStateKey, carrierqueue.InitStateDone, 0)
|
||||
c.logger.Info("按运营商账户调度初始化完成，开始领取", zap.Int64("cards", written), zap.Duration("duration", time.Since(started)),
|
||||
zap.String("metric", "polling.carrier.initialized"))
|
||||
}
|
||||
|
||||
// syncConsumers 按当前运营商账户订阅卡任务 topic，max_in_flight 取账户调度参数。
|
||||
func (c *carrierSchedulerRuntime) syncConsumers(ctx context.Context) {
|
||||
if c.consumers == nil || c.dispatcher == nil {
|
||||
return
|
||||
}
|
||||
desired := map[string]int{}
|
||||
for _, carrierID := range c.dispatcher.CarrierIDs(ctx) {
|
||||
desired[constants.NSQCardTopic(carrierID)] = c.dispatcher.ParamsFor(carrierID).CardMaxInFlight
|
||||
}
|
||||
c.consumers.Sync(desired)
|
||||
}
|
||||
|
||||
func (c *carrierSchedulerRuntime) grace() time.Duration {
|
||||
if c.cfg.ShutdownGrace > 0 {
|
||||
return c.cfg.ShutdownGrace
|
||||
}
|
||||
return 30 * time.Second
|
||||
}
|
||||
|
||||
// shutdown 优雅退出：停止订阅，宽限期内等待执行中的卡任务完成；未完成的由租约到期后重新领取。
|
||||
func (c *carrierSchedulerRuntime) shutdown() {
|
||||
if c.consumers != nil {
|
||||
c.consumers.StopAll(c.grace())
|
||||
}
|
||||
if c.producer != nil {
|
||||
c.producer.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
// InitProgress 返回新结构初始化进度。
|
||||
func (c *carrierSchedulerRuntime) InitProgress() polling.InitProgress {
|
||||
return c.initializer.Progress()
|
||||
}
|
||||
@@ -45,6 +45,7 @@ import (
|
||||
wecomInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/wecom"
|
||||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||||
"github.com/break/junhong_cmp_fiber/internal/polling"
|
||||
"github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue"
|
||||
iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
|
||||
pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling"
|
||||
refundSvc "github.com/break/junhong_cmp_fiber/internal/service/refund"
|
||||
@@ -102,6 +103,12 @@ type workerRuntime struct {
|
||||
assetAutoRenewalService *assetAutoRenewalApp.Service
|
||||
// gatewayLimit 是卡网关运营商账户限流组件（限流、统计、停复机记录）。
|
||||
gatewayLimit *gatewaylimit.Runtime
|
||||
// 按运营商账户调度：调度参数、账户队列存储与模式接缝（模式为 carrier 时各入口写新结构）。
|
||||
carrierSettings *carrierqueue.SettingsProvider
|
||||
carrierStore *carrierqueue.Store
|
||||
carrierScheduling *polling.CarrierScheduling
|
||||
// carrierReinit 在 carrier 模式下轮询配置变更时请求重新初始化新结构。
|
||||
carrierReinit func()
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -171,6 +178,12 @@ func runWorker(cfg *config.Config) {
|
||||
|
||||
taskHandler := createTaskHandler(runtime, appLogger)
|
||||
taskHandler.RegisterHandlers()
|
||||
carrierRuntime := startCarrierScheduler(ctx, cfg, runtime, taskHandler.PollingSteps(), runsSingletonModules(cfg.Worker.Role), func() {
|
||||
if pollingInitializer != nil {
|
||||
pollingInitializer.Restart(ctx)
|
||||
}
|
||||
}, appLogger)
|
||||
runtime.carrierReinit = carrierRuntime.RequestReinit
|
||||
registerWeComApprovalTasks(taskHandler.GetMux(), runtime, cfg, appLogger)
|
||||
registerAgentRechargeRecoveryTask(taskHandler.GetMux(), runtime, appLogger)
|
||||
registerRefundChannelRecoveryTask(taskHandler.GetMux(), runtime, appLogger)
|
||||
@@ -379,6 +392,7 @@ func initWorkerRuntime(ctx context.Context, cfg *config.Config, appLogger *zap.L
|
||||
// 自动续费的复机执行同样复用既有停复机单一事实源。
|
||||
assetAutoRenewalService: workerResult.Services.AssetAutoRenewal,
|
||||
}
|
||||
newCarrierSchedulerParts(runtime, appLogger)
|
||||
registerNotificationOutboxConsumer(runtime, appLogger)
|
||||
registerWalletOutboxConsumer(runtime, appLogger)
|
||||
registerCardObservationOutboxConsumer(runtime, appLogger)
|
||||
@@ -842,6 +856,13 @@ func startPollingInitializer(ctx context.Context, runtime *workerRuntime, appLog
|
||||
pollingInitializer.StartBackground(ctx)
|
||||
|
||||
runtime.pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) {
|
||||
if runtime.carrierScheduling.Active(ctx) {
|
||||
// 按运营商账户调度时按新配置重新初始化新结构，旧分片队列在回退时再重建。
|
||||
if runtime.carrierReinit != nil {
|
||||
runtime.carrierReinit()
|
||||
}
|
||||
return
|
||||
}
|
||||
if hasConfigs {
|
||||
appLogger.Info("轮询配置已变更，触发队列重新初始化",
|
||||
zap.Bool("had_configs", hadConfigs))
|
||||
@@ -894,6 +915,7 @@ func startPollingScheduler(
|
||||
dataResetHandler,
|
||||
)
|
||||
pollingScheduler.SetInitializer(pollingInitializer)
|
||||
pollingScheduler.SetCarrierScheduling(runtime.carrierScheduling)
|
||||
|
||||
if err := pollingScheduler.Start(ctx); err != nil {
|
||||
appLogger.Error("启动轮询调度器失败", zap.Error(err))
|
||||
|
||||
Reference in New Issue
Block a user