diff --git a/cmd/worker/main.go b/cmd/worker/main.go index f9573c4..1592d2d 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -9,12 +9,14 @@ import ( "syscall" "time" + "github.com/bytedance/sonic" "github.com/hibiken/asynq" "github.com/redis/go-redis/v9" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/bootstrap" "github.com/break/junhong_cmp_fiber/internal/gateway" + "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/polling" iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" "github.com/break/junhong_cmp_fiber/internal/store/postgres" @@ -34,6 +36,7 @@ const ( workerModulePollingInitializer = "polling_initializer" workerModulePollingScheduler = "polling_scheduler" workerModuleAsynqScheduler = "asynq_scheduler" + importRescueLimit = 500 // 启动补偿单次扫描的最大导入任务数 ) // workerModuleStatus 描述当前 Worker 角色下的模块启停计划。 @@ -121,6 +124,7 @@ func runWorker(cfg *config.Config) { taskHandler := createTaskHandler(runtime, appLogger) taskHandler.RegisterHandlers() + rescuePendingImportTasks(ctx, runtime, appLogger) appLogger.Info("Worker 服务器配置完成", zap.Int("concurrency", cfg.Queue.Concurrency), @@ -382,18 +386,36 @@ func startAsynqScheduler(cfg *config.Config, redisAddr string, appLogger *zap.Lo // registerAsynqScheduleTasks 注册 Worker 入口需要的全部定时任务。 func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler) error { - if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask(constants.TaskTypeOrderExpire, nil)); err != nil { + if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask( + constants.TaskTypeOrderExpire, + nil, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeOrderExpire)), + )); err != nil { return fmt.Errorf("注册订单超时定时任务失败: %w", err) } - if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask(constants.TaskTypeAlertCheck, nil)); err != nil { + if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask( + constants.TaskTypeAlertCheck, + nil, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeAlertCheck)), + )); err != nil { return fmt.Errorf("注册告警检查定时任务失败: %w", err) } - if _, err := asynqScheduler.Register("0 2 * * *", asynq.NewTask(constants.TaskTypeDataCleanup, nil)); err != nil { + if _, err := asynqScheduler.Register("0 2 * * *", asynq.NewTask( + constants.TaskTypeDataCleanup, + nil, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeDataCleanup)), + )); err != nil { return fmt.Errorf("注册数据清理定时任务失败: %w", err) } if _, err := asynqScheduler.Register( "0 2 * * *", - asynq.NewTask(constants.TaskTypeDailyTrafficFlush, nil, asynq.MaxRetry(3), asynq.Timeout(5*time.Minute)), + asynq.NewTask( + constants.TaskTypeDailyTrafficFlush, + nil, + asynq.MaxRetry(3), + asynq.Timeout(5*time.Minute), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeDailyTrafficFlush)), + ), ); err != nil { return fmt.Errorf("注册每日流量落盘定时任务失败: %w", err) } @@ -417,6 +439,80 @@ func createTaskHandler(runtime *workerRuntime, appLogger *zap.Logger) *queue.Han ) } +// rescuePendingImportTasks 将历史遗留的待处理导入任务重新投递到独立导入队列。 +func rescuePendingImportTasks(ctx context.Context, runtime *workerRuntime, appLogger *zap.Logger) { + rescuePendingIotCardImportTasks(ctx, runtime.db, runtime.asynqClient, appLogger) + rescuePendingDeviceImportTasks(ctx, runtime.db, runtime.asynqClient, appLogger) +} + +// rescuePendingIotCardImportTasks 补偿仍停留在待处理状态的 IoT 卡导入任务。 +func rescuePendingIotCardImportTasks(ctx context.Context, db *gorm.DB, asynqClient *asynq.Client, appLogger *zap.Logger) { + var importTasks []model.IotCardImportTask + if err := db.WithContext(ctx). + Where("status = ?", model.ImportTaskStatusPending). + Limit(importRescueLimit). + Find(&importTasks).Error; err != nil { + appLogger.Warn("扫描待补偿 IoT 卡导入任务失败", zap.Error(err)) + return + } + + for _, importTask := range importTasks { + payload := task.IotCardImportPayload{TaskID: importTask.ID} + enqueueImportRescueTask(ctx, asynqClient, constants.TaskTypeIotCardImport, payload, importTask.ID, appLogger) + } +} + +// rescuePendingDeviceImportTasks 补偿仍停留在待处理状态的设备导入任务。 +func rescuePendingDeviceImportTasks(ctx context.Context, db *gorm.DB, asynqClient *asynq.Client, appLogger *zap.Logger) { + var importTasks []model.DeviceImportTask + if err := db.WithContext(ctx). + Where("status = ?", model.ImportTaskStatusPending). + Limit(importRescueLimit). + Find(&importTasks).Error; err != nil { + appLogger.Warn("扫描待补偿设备导入任务失败", zap.Error(err)) + return + } + + for _, importTask := range importTasks { + payload := task.DeviceImportPayload{TaskID: importTask.ID} + enqueueImportRescueTask(ctx, asynqClient, constants.TaskTypeDeviceImport, payload, importTask.ID, appLogger) + } +} + +// enqueueImportRescueTask 将补偿任务提交到任务类型对应的独立队列。 +func enqueueImportRescueTask(ctx context.Context, asynqClient *asynq.Client, taskType string, payload any, taskID uint, appLogger *zap.Logger) { + payloadBytes, err := sonic.Marshal(payload) + if err != nil { + appLogger.Warn("序列化导入补偿任务载荷失败", + zap.String("task_type", taskType), + zap.Uint("task_id", taskID), + zap.Error(err)) + return + } + + queueName := constants.QueueForTaskType(taskType) + taskMessage := asynq.NewTask( + taskType, + payloadBytes, + asynq.Queue(queueName), + asynq.TaskID(fmt.Sprintf("import-rescue:%s:%d", taskType, taskID)), + asynq.Unique(30*time.Minute), + ) + if _, err := asynqClient.EnqueueContext(ctx, taskMessage); err != nil { + appLogger.Warn("提交导入补偿任务失败", + zap.String("task_type", taskType), + zap.String("queue", queueName), + zap.Uint("task_id", taskID), + zap.Error(err)) + return + } + + appLogger.Info("导入补偿任务已提交", + zap.String("task_type", taskType), + zap.String("queue", queueName), + zap.Uint("task_id", taskID)) +} + // shutdownWorker 按当前实例实际启动过的模块执行优雅关闭。 func shutdownWorker( cancel context.CancelFunc, diff --git a/internal/polling/package_activation_handler.go b/internal/polling/package_activation_handler.go index 2d72b8d..1e39e1b 100644 --- a/internal/polling/package_activation_handler.go +++ b/internal/polling/package_activation_handler.go @@ -484,7 +484,7 @@ func (h *PackageActivationHandler) enqueueActivationTask(ctx context.Context, pa task := asynq.NewTask(constants.TaskTypePackageQueueActivation, payloadBytes, asynq.MaxRetry(3), asynq.Timeout(30*time.Second), - asynq.Queue(constants.QueueDefault), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageQueueActivation)), ) _, err = h.queueClient.Enqueue(task) diff --git a/internal/polling/scheduler.go b/internal/polling/scheduler.go index 81c6d3c..1ec2436 100644 --- a/internal/polling/scheduler.go +++ b/internal/polling/scheduler.go @@ -263,7 +263,7 @@ func (s *Scheduler) enqueueBatch(ctx context.Context, taskType string, cardIDs [ task := asynq.NewTask(taskType, payloadBytes, asynq.MaxRetry(0), asynq.Timeout(30*time.Second), - asynq.Queue(constants.QueueDefault), + asynq.Queue(constants.QueueForTaskType(taskType)), ) if _, err := s.queueClient.Enqueue(task); err != nil { s.logger.Error("提交任务失败，回退至分片队列防止卡永久丢失", diff --git a/internal/service/device_import/service.go b/internal/service/device_import/service.go index a0a7080..cae9e0c 100644 --- a/internal/service/device_import/service.go +++ b/internal/service/device_import/service.go @@ -13,6 +13,7 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/errors" "github.com/break/junhong_cmp_fiber/pkg/middleware" "github.com/break/junhong_cmp_fiber/pkg/queue" + "github.com/hibiken/asynq" "gorm.io/gorm" ) @@ -70,7 +71,12 @@ func (s *Service) CreateImportTask(ctx context.Context, req *dto.ImportDeviceReq } payload := DeviceImportPayload{TaskID: task.ID} - err := s.queueClient.EnqueueTask(ctx, constants.TaskTypeDeviceImport, payload) + err := s.queueClient.EnqueueTask( + ctx, + constants.TaskTypeDeviceImport, + payload, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeDeviceImport)), + ) if err != nil { s.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "任务入队失败: "+err.Error()) appErr := errors.Wrap(errors.CodeInternalError, err, "任务入队失败") diff --git a/internal/service/email/service.go b/internal/service/email/service.go index 18ee2cf..8749bab 100644 --- a/internal/service/email/service.go +++ b/internal/service/email/service.go @@ -10,7 +10,6 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" "github.com/break/junhong_cmp_fiber/pkg/queue" - "github.com/bytedance/sonic" "github.com/hibiken/asynq" "go.uber.org/zap" ) @@ -39,21 +38,12 @@ func (s *Service) SendWelcomeEmail(ctx context.Context, userID uint, email strin Body: "感谢您注册我们的服务！我们很高兴为您提供服务。", } - payloadBytes, err := sonic.Marshal(payload) - if err != nil { - s.logger.Error("序列化邮件任务载荷失败", - zap.Uint("user_id", userID), - zap.String("email", email), - zap.Error(err)) - return errors.Wrap(errors.CodeInternalError, err, "序列化邮件任务载荷失败") - } - // 提交任务到队列 - err = s.queueClient.EnqueueTask( + err := s.queueClient.EnqueueTask( ctx, constants.TaskTypeEmailSend, - payloadBytes, - asynq.Queue(constants.QueueDefault), + payload, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeEmailSend)), asynq.MaxRetry(constants.DefaultRetryMax), ) @@ -82,20 +72,12 @@ func (s *Service) SendPasswordResetEmail(ctx context.Context, email string, rese Body: fmt.Sprintf("您的密码重置令牌是: %s\n此令牌将在 1 小时后过期。", resetToken), } - payloadBytes, err := sonic.Marshal(payload) - if err != nil { - s.logger.Error("序列化密码重置邮件任务载荷失败", - zap.String("email", email), - zap.Error(err)) - return errors.Wrap(errors.CodeInternalError, err, "序列化密码重置邮件任务载荷失败") - } - - // 提交任务到队列（高优先级） - err = s.queueClient.EnqueueTask( + // 提交任务到独立邮件队列 + err := s.queueClient.EnqueueTask( ctx, constants.TaskTypeEmailSend, - payloadBytes, - asynq.Queue(constants.QueueCritical), // 密码重置使用高优先级队列 + payload, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeEmailSend)), asynq.MaxRetry(constants.DefaultRetryMax), ) @@ -122,20 +104,12 @@ func (s *Service) SendNotificationEmail(ctx context.Context, to string, subject Body: body, } - payloadBytes, err := sonic.Marshal(payload) - if err != nil { - s.logger.Error("序列化通知邮件任务载荷失败", - zap.String("to", to), - zap.Error(err)) - return errors.Wrap(errors.CodeInternalError, err, "序列化通知邮件任务载荷失败") - } - - // 提交任务到队列（低优先级） - err = s.queueClient.EnqueueTask( + // 提交任务到独立邮件队列 + err := s.queueClient.EnqueueTask( ctx, constants.TaskTypeEmailSend, - payloadBytes, - asynq.Queue(constants.QueueLow), // 通知邮件使用低优先级队列 + payload, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeEmailSend)), asynq.MaxRetry(constants.DefaultRetryMax), ) diff --git a/internal/service/export_task/service.go b/internal/service/export_task/service.go index 52060fc..37f6d26 100644 --- a/internal/service/export_task/service.go +++ b/internal/service/export_task/service.go @@ -126,9 +126,9 @@ func (s *Service) CreateTask(ctx context.Context, req *dto.CreateExportTaskReque ctx, constants.TaskTypeExportDispatch, dispatchPayload{TaskID: task.ID}, - asynq.Queue(constants.QueueDefault), asynq.MaxRetry(constants.ExportDispatchRetryMax), asynq.Timeout(constants.ExportDispatchTaskTimeout), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeExportDispatch)), ); err != nil { _ = s.taskStore.MarkFailed(ctx, task.ID, userID, "导出任务入队失败") return nil, errors.Wrap(errors.CodeTaskQueueError, err, "导出任务入队失败") diff --git a/internal/service/iot_card_import/service.go b/internal/service/iot_card_import/service.go index 49ca026..2445331 100644 --- a/internal/service/iot_card_import/service.go +++ b/internal/service/iot_card_import/service.go @@ -13,6 +13,7 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/errors" "github.com/break/junhong_cmp_fiber/pkg/middleware" "github.com/break/junhong_cmp_fiber/pkg/queue" + "github.com/hibiken/asynq" "gorm.io/gorm" ) @@ -108,7 +109,12 @@ func (s *Service) CreateImportTask(ctx context.Context, req *dto.ImportIotCardRe } payload := IotCardImportPayload{TaskID: task.ID} - err = s.queueClient.EnqueueTask(ctx, constants.TaskTypeIotCardImport, payload) + err = s.queueClient.EnqueueTask( + ctx, + constants.TaskTypeIotCardImport, + payload, + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeIotCardImport)), + ) if err != nil { s.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "任务入队失败: "+err.Error()) appErr := errors.Wrap(errors.CodeInternalError, err, "任务入队失败") diff --git a/internal/service/recharge_order/service.go b/internal/service/recharge_order/service.go index e57b636..613e07f 100644 --- a/internal/service/recharge_order/service.go +++ b/internal/service/recharge_order/service.go @@ -169,8 +169,8 @@ func (s *Service) HandlePaymentCallback(ctx context.Context, paymentNo string, p if len(linkedIDs) > 0 { taskPayload := task.AutoPurchasePayload{RechargeOrderID: rechargeOrder.ID} if err := s.queueClient.EnqueueTask(ctx, constants.TaskTypeAutoPurchaseAfterRecharge, taskPayload, - asynq.Queue(constants.QueueDefault), asynq.MaxRetry(3), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeAutoPurchaseAfterRecharge)), ); err != nil { s.logger.Error("自动购包任务入队失败", zap.Uint("recharge_order_id", rechargeOrder.ID), diff --git a/internal/task/auto_purchase.go b/internal/task/auto_purchase.go index 63c3125..bd6db45 100644 --- a/internal/task/auto_purchase.go +++ b/internal/task/auto_purchase.go @@ -271,8 +271,8 @@ func (h *AutoPurchaseHandler) ProcessTask(ctx context.Context, task *asynq.Task) zap.Error(marshalErr)) } else { commissionTask := asynq.NewTask(constants.TaskTypeCommission, payloadBytes, - asynq.Queue(constants.QueueLow), asynq.MaxRetry(3), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeCommission)), ) if _, enqueueErr := h.asynqClient.EnqueueContext(ctx, commissionTask); enqueueErr != nil { h.logger.Warn("自动购包后提交佣金任务失败", @@ -296,7 +296,7 @@ func NewAutoPurchaseTask(rechargeOrderID uint) (*asynq.Task, error) { return asynq.NewTask(constants.TaskTypeAutoPurchaseAfterRecharge, payloadBytes, asynq.MaxRetry(3), asynq.Timeout(2*time.Minute), - asynq.Queue(constants.QueueDefault), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeAutoPurchaseAfterRecharge)), ), nil } diff --git a/internal/task/export_common.go b/internal/task/export_common.go index 714a58e..3741c78 100644 --- a/internal/task/export_common.go +++ b/internal/task/export_common.go @@ -50,6 +50,7 @@ func enqueueTask(ctx context.Context, client *asynq.Client, taskType string, pay return err } + opts = append(opts, asynq.Queue(constants.QueueForTaskType(taskType))) task := asynq.NewTask(taskType, bytes, opts...) _, err = client.EnqueueContext(ctx, task) return err diff --git a/internal/task/export_dispatch.go b/internal/task/export_dispatch.go index 1379673..372928f 100644 --- a/internal/task/export_dispatch.go +++ b/internal/task/export_dispatch.go @@ -215,9 +215,9 @@ func (h *ExportDispatchHandler) enqueueShardTasks(ctx context.Context, task *mod h.asynqClient, constants.TaskTypeExportShard, payload, - asynq.Queue(constants.QueueLow), asynq.MaxRetry(constants.ExportShardRetryMax), asynq.Timeout(constants.ExportShardTaskTimeout), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeExportShard)), ) if err != nil { h.logger.Error("导出分片任务入队失败", @@ -237,10 +237,10 @@ func (h *ExportDispatchHandler) enqueueFinalizeTask(ctx context.Context, taskID h.asynqClient, constants.TaskTypeExportFinalize, ExportFinalizePayload{TaskID: taskID}, - asynq.Queue(constants.QueueDefault), asynq.MaxRetry(constants.ExportFinalizeRetryMax), asynq.Timeout(constants.ExportFinalizeTaskTimeout), asynq.ProcessIn(5*time.Second), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeExportFinalize)), ) } diff --git a/internal/task/export_shard.go b/internal/task/export_shard.go index 3cd0f30..88d308b 100644 --- a/internal/task/export_shard.go +++ b/internal/task/export_shard.go @@ -175,9 +175,9 @@ func (h *ExportShardHandler) enqueueFinalize(ctx context.Context, taskID uint) e h.asynqClient, constants.TaskTypeExportFinalize, ExportFinalizePayload{TaskID: taskID}, - asynq.Queue(constants.QueueDefault), asynq.MaxRetry(constants.ExportFinalizeRetryMax), asynq.Timeout(constants.ExportFinalizeTaskTimeout), asynq.ProcessIn(2*time.Second), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypeExportFinalize)), ) } diff --git a/internal/task/polling_realname_handler.go b/internal/task/polling_realname_handler.go index abc03d3..dff6a68 100644 --- a/internal/task/polling_realname_handler.go +++ b/internal/task/polling_realname_handler.go @@ -206,7 +206,7 @@ func (h *PollingRealnameHandler) triggerFirstRealnameActivation(ctx context.Cont task := asynq.NewTask(constants.TaskTypePackageFirstActivation, payloadBytes, asynq.MaxRetry(3), asynq.Timeout(30*time.Second), - asynq.Queue(constants.QueueDefault), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageFirstActivation)), ) if _, err := h.asynqClient.EnqueueContext(ctx, task); err != nil { h.base.logger.Warn("提交首次实名激活任务失败", zap.Uint("card_id", cardID), zap.Error(err)) @@ -244,7 +244,7 @@ func (h *PollingRealnameHandler) triggerDeviceRealnameActivation(ctx context.Con task := asynq.NewTask(constants.TaskTypePackageFirstActivation, payloadBytes, asynq.MaxRetry(3), asynq.Timeout(30*time.Second), - asynq.Queue(constants.QueueDefault), + asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageFirstActivation)), ) if _, err := h.asynqClient.EnqueueContext(ctx, task); err != nil { h.base.logger.Warn("提交设备实名激活任务失败", diff --git a/pkg/config/config.go b/pkg/config/config.go index bc4343e..fa5d888 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -70,7 +70,7 @@ type DatabaseConfig struct { // QueueConfig 任务队列配置 type QueueConfig struct { Concurrency int `mapstructure:"concurrency"` // Worker 并发数（默认：10） - Queues map[string]int `mapstructure:"queues"` // 队列优先级配置（队列名 -> 权重） + Queues map[string]int `mapstructure:"queues"` // Asynq 队列取任务权重（不是业务并发上限） RetryMax int `mapstructure:"retry_max"` // 最大重试次数（默认：5） Timeout time.Duration `mapstructure:"timeout"` // 任务超时时间（默认：10m） } diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index 530cb80..83c8630 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -172,14 +172,125 @@ const ( // 队列配置常量 const ( - QueueCritical = "critical" // 关键任务队列 - QueueDefault = "default" // 默认队列 - QueueLow = "low" // 低优先级队列 - DefaultRetryMax = 5 - DefaultTimeout = 10 * time.Minute - DefaultConcurrency = 10 + QueueCritical = "critical" // 旧版关键任务队列（兼容历史任务） + QueueDefault = "default" // 旧版默认队列（兼容历史任务） + QueueLow = "low" // 旧版低优先级队列（兼容历史任务） + + QueueEmailSend = TaskTypeEmailSend // 发送邮件任务队列 + QueueCommission = TaskTypeCommission // 分佣计算任务队列 + QueueIotCardImport = TaskTypeIotCardImport // IoT 卡导入任务队列 + QueueDeviceImport = TaskTypeDeviceImport // 设备导入任务队列 + QueueExportDispatch = TaskTypeExportDispatch // 导出分发任务队列 + QueueExportShard = TaskTypeExportShard // 导出分片任务队列 + QueueExportFinalize = TaskTypeExportFinalize // 导出收尾任务队列 + QueueCommissionStatsUpdate = TaskTypeCommissionStatsUpdate // 佣金统计更新任务队列 + QueueCommissionStatsSync = TaskTypeCommissionStatsSync // 佣金统计同步任务队列 + QueueCommissionStatsArchive = TaskTypeCommissionStatsArchive // 佣金统计归档任务队列 + QueuePollingRealname = TaskTypePollingRealname // 实名轮询任务队列 + QueuePollingCarddata = TaskTypePollingCarddata // 流量轮询任务队列 + QueuePollingPackage = TaskTypePollingPackage // 套餐轮询任务队列 + QueuePollingProtect = TaskTypePollingProtect // 保护期轮询任务队列 + QueuePollingCardStatus = TaskTypePollingCardStatus // 卡状态轮询任务队列 + QueuePackageFirstActivation = TaskTypePackageFirstActivation // 首次实名激活任务队列 + QueuePackageQueueActivation = TaskTypePackageQueueActivation // 主套餐排队激活任务队列 + QueuePackageDataReset = TaskTypePackageDataReset // 套餐流量重置任务队列 + QueueOrderExpire = TaskTypeOrderExpire // 订单超时取消任务队列 + QueueAutoPurchase = TaskTypeAutoPurchaseAfterRecharge // 充值后自动购包任务队列 + QueueAlertCheck = TaskTypeAlertCheck // 告警检查任务队列 + QueueDataCleanup = TaskTypeDataCleanup // 数据清理任务队列 + QueueDailyTrafficFlush = TaskTypeDailyTrafficFlush // 每日流量落盘任务队列 + + DefaultRetryMax = 5 // 默认任务最大重试次数 + DefaultTimeout = 10 * time.Minute // 默认任务超时时间 + DefaultConcurrency = 10 // 默认 Worker 并发数 ) +// QueueForTaskType 根据任务类型返回独立队列名。 +func QueueForTaskType(taskType string) string { + switch taskType { + case TaskTypeEmailSend: + return QueueEmailSend + case TaskTypeCommission: + return QueueCommission + case TaskTypeIotCardImport: + return QueueIotCardImport + case TaskTypeDeviceImport: + return QueueDeviceImport + case TaskTypeExportDispatch: + return QueueExportDispatch + case TaskTypeExportShard: + return QueueExportShard + case TaskTypeExportFinalize: + return QueueExportFinalize + case TaskTypeCommissionStatsUpdate: + return QueueCommissionStatsUpdate + case TaskTypeCommissionStatsSync: + return QueueCommissionStatsSync + case TaskTypeCommissionStatsArchive: + return QueueCommissionStatsArchive + case TaskTypePollingRealname: + return QueuePollingRealname + case TaskTypePollingCarddata: + return QueuePollingCarddata + case TaskTypePollingPackage: + return QueuePollingPackage + case TaskTypePollingProtect: + return QueuePollingProtect + case TaskTypePollingCardStatus: + return QueuePollingCardStatus + case TaskTypePackageFirstActivation: + return QueuePackageFirstActivation + case TaskTypePackageQueueActivation: + return QueuePackageQueueActivation + case TaskTypePackageDataReset: + return QueuePackageDataReset + case TaskTypeOrderExpire: + return QueueOrderExpire + case TaskTypeAutoPurchaseAfterRecharge: + return QueueAutoPurchase + case TaskTypeAlertCheck: + return QueueAlertCheck + case TaskTypeDataCleanup: + return QueueDataCleanup + case TaskTypeDailyTrafficFlush: + return QueueDailyTrafficFlush + default: + return QueueDefault + } +} + +// DefaultTaskQueueWeights 返回 Worker 默认监听的全部任务队列及其权重。 +func DefaultTaskQueueWeights() map[string]int { + return map[string]int{ + QueueCritical: 6, + QueueIotCardImport: 6, + QueueDeviceImport: 6, + QueueAutoPurchase: 5, + QueuePackageFirstActivation: 5, + QueuePackageQueueActivation: 5, + QueueOrderExpire: 4, + QueueEmailSend: 4, + QueueExportDispatch: 4, + QueueExportFinalize: 4, + QueueCommission: 3, + QueueAlertCheck: 3, + QueueExportShard: 2, + QueueCommissionStatsUpdate: 2, + QueueCommissionStatsSync: 2, + QueueCommissionStatsArchive: 2, + QueuePollingRealname: 1, + QueuePollingCarddata: 1, + QueuePollingPackage: 1, + QueuePollingProtect: 1, + QueuePollingCardStatus: 1, + QueuePackageDataReset: 1, + QueueDataCleanup: 1, + QueueDailyTrafficFlush: 1, + QueueDefault: 1, + QueueLow: 1, + } +} + // 导出任务相关常量 const ( ExportTaskSceneDevice = "device" // 导出场景：设备 diff --git a/pkg/queue/client.go b/pkg/queue/client.go index cf046c8..5341ccb 100644 --- a/pkg/queue/client.go +++ b/pkg/queue/client.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/break/junhong_cmp_fiber/pkg/config" + "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/bytedance/sonic" "github.com/hibiken/asynq" "github.com/redis/go-redis/v9" @@ -38,6 +39,12 @@ func NewClient(redisClient *redis.Client, logger *zap.Logger) *Client { // payload 必须传入 struct 或 map，禁止传入 []byte。 // 传入 []byte 会导致 sonic.Marshal 将其 base64 编码，handler 解析时类型不匹配。 func (c *Client) EnqueueTask(ctx context.Context, taskType string, payload interface{}, opts ...asynq.Option) error { + if _, ok := payload.([]byte); ok { + c.logger.Error("任务载荷类型错误，禁止传入 []byte", + zap.String("task_type", taskType)) + return fmt.Errorf("task payload must be struct or map, got []byte") + } + // 内部统一序列化，调用方无需预先 Marshal payloadBytes, err := sonic.Marshal(payload) if err != nil { @@ -47,6 +54,8 @@ func (c *Client) EnqueueTask(ctx context.Context, taskType string, payload inter return fmt.Errorf("failed to marshal task payload: %w", err) } + opts = append(opts, asynq.Queue(constants.QueueForTaskType(taskType))) + // 创建任务 task := asynq.NewTask(taskType, payloadBytes, opts...) @@ -78,13 +87,12 @@ func (c *Client) Close() error { // ParseQueueConfig 解析队列配置为 Asynq 格式 func ParseQueueConfig(cfg *config.QueueConfig) map[string]int { + queues := constants.DefaultTaskQueueWeights() if cfg.Queues != nil && len(cfg.Queues) > 0 { - return cfg.Queues - } - // 默认队列优先级 - return map[string]int{ - "critical": 6, - "default": 3, - "low": 1, + // 配置只覆盖权重；任务队列全集由代码兜底，避免漏监听新队列。 + for queueName, weight := range cfg.Queues { + queues[queueName] = weight + } } + return queues }