From 3a2e3f25711fa1cd2af31e690b7aa69dead7c14b Mon Sep 17 00:00:00 2001 From: huang Date: Wed, 6 May 2026 14:41:14 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=85=E6=B5=85=E5=8D=87=E7=BA=A7=E4=B8=80?= =?UTF-8?q?=E4=B8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/worker/main.go | 376 +++++++++--- docs/environment-variables.md | 74 +++ docs/polling-system/performance-tuning.md | 25 +- .../.openspec.yaml | 2 + .../2026-05-06-split-worker-roles/design.md | 99 ++++ .../2026-05-06-split-worker-roles/proposal.md | 63 ++ .../specs/embedded-config/spec.md | 45 ++ .../specs/worker-role-management/spec.md | 125 ++++ .../2026-05-06-split-worker-roles/tasks.md | 38 ++ openspec/specs/embedded-config/spec.md | 28 +- openspec/specs/worker-role-management/spec.md | 130 +++++ pkg/config/config.go | 24 + pkg/config/defaults/config.yaml | 4 + pkg/config/loader.go | 2 + pkg/constants/worker.go | 11 + worker-多实例升级方案.md | 550 ++++++++++++++++++ 16 files changed, 1490 insertions(+), 106 deletions(-) create mode 100644 openspec/changes/archive/2026-05-06-split-worker-roles/.openspec.yaml create mode 100644 openspec/changes/archive/2026-05-06-split-worker-roles/design.md create mode 100644 openspec/changes/archive/2026-05-06-split-worker-roles/proposal.md create mode 100644 openspec/changes/archive/2026-05-06-split-worker-roles/specs/embedded-config/spec.md create mode 100644 openspec/changes/archive/2026-05-06-split-worker-roles/specs/worker-role-management/spec.md create mode 100644 openspec/changes/archive/2026-05-06-split-worker-roles/tasks.md create mode 100644 openspec/specs/worker-role-management/spec.md create mode 100644 pkg/constants/worker.go create mode 100644 worker-多实例升级方案.md diff --git a/cmd/worker/main.go b/cmd/worker/main.go index beebc50..f9573c4 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "fmt" "os" "os/signal" "strconv" @@ -25,14 +26,50 @@ import ( "github.com/break/junhong_cmp_fiber/pkg/logger" "github.com/break/junhong_cmp_fiber/pkg/queue" "github.com/break/junhong_cmp_fiber/pkg/storage" + "gorm.io/gorm" ) +const ( + workerModuleQueueServer = "queue_server" + workerModulePollingInitializer = "polling_initializer" + workerModulePollingScheduler = "polling_scheduler" + workerModuleAsynqScheduler = "asynq_scheduler" +) + +// workerModuleStatus 描述当前 Worker 角色下的模块启停计划。 +type workerModuleStatus struct { + enabled []string + disabled []string +} + +// workerRuntime 保存所有角色共享的 Worker 运行时依赖。 +type workerRuntime struct { + redisAddr string + redisClient *redis.Client + db *gorm.DB + storageSvc *storage.Service + gatewayClient *gateway.Client + asynqClient *asynq.Client + workerResult *bootstrap.WorkerBootstrapResult + workerServer *queue.Server + pollingConfigMgr *polling.PollingConfigManager + pollingQueueMgr *polling.PollingQueueManager + pollingIotCardStore *postgres.IotCardStore + pollingBase *task.PollingBase + lifecycleSvc *polling.PollingLifecycleService +} + func main() { cfg, err := config.Load() if err != nil { panic("加载配置失败: " + err.Error()) } + runWorker(cfg) +} + +// runWorker 编排 Worker 进程启动、模块启停和优雅关闭流程。 +func runWorker(cfg *config.Config) { if _, err := pkgBootstrap.EnsureDirectories(cfg, nil); err != nil { panic("初始化目录失败: " + err.Error()) } @@ -62,9 +99,90 @@ func main() { }() appLogger := logger.GetAppLogger() - appLogger.Info("Worker 服务启动中...") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() - // 连接 Redis + moduleStatus := buildWorkerModuleStatus(cfg.Worker.Role) + appLogger.Info("Worker 服务启动中...") + logWorkerRole(appLogger, cfg.Worker.Role, cfg.Worker.InstanceName, moduleStatus) + + runtime := initWorkerRuntime(ctx, cfg, appLogger) + defer runtime.close(appLogger) + + var pollingInitializer *polling.PollingInitializer + var pollingScheduler *polling.Scheduler + var asynqScheduler *asynq.Scheduler + + if runsSingletonModules(cfg.Worker.Role) { + pollingInitializer = startPollingInitializer(ctx, runtime, appLogger) + pollingScheduler = startPollingScheduler(ctx, runtime, pollingInitializer, appLogger) + asynqScheduler = startAsynqScheduler(cfg, runtime.redisAddr, appLogger) + } + + taskHandler := createTaskHandler(runtime, appLogger) + taskHandler.RegisterHandlers() + + appLogger.Info("Worker 服务器配置完成", + zap.Int("concurrency", cfg.Queue.Concurrency), + zap.Any("queues", cfg.Queue.Queues)) + + quit := make(chan os.Signal, 1) + signal.Notify(quit, os.Interrupt, syscall.SIGTERM) + defer signal.Stop(quit) + + go func() { + if err := runtime.workerServer.Run(taskHandler.GetMux()); err != nil { + appLogger.Fatal("Worker 服务器运行失败", zap.Error(err)) + } + }() + + appLogger.Info("Worker 服务器已启动") + <-quit + + shutdownWorker(cancel, appLogger, runtime.workerServer, pollingInitializer, pollingScheduler, asynqScheduler) +} + +// buildWorkerModuleStatus 根据角色生成启用/禁用模块列表，供启动日志和人工验收使用。 +func buildWorkerModuleStatus(role string) workerModuleStatus { + status := workerModuleStatus{ + enabled: []string{workerModuleQueueServer}, + } + + if runsSingletonModules(role) { + status.enabled = append( + status.enabled, + workerModulePollingInitializer, + workerModulePollingScheduler, + workerModuleAsynqScheduler, + ) + return status + } + + status.disabled = []string{ + workerModulePollingInitializer, + workerModulePollingScheduler, + workerModuleAsynqScheduler, + } + return status +} + +// logWorkerRole 输出当前实例角色、实例名以及模块启停计划。 +func logWorkerRole(appLogger *zap.Logger, role string, instanceName string, moduleStatus workerModuleStatus) { + appLogger.Info("Worker 角色已确认", + zap.String("role", role), + zap.String("instance_name", instanceName)) + appLogger.Info("Worker 模块启停计划", + zap.Strings("enabled_modules", moduleStatus.enabled), + zap.Strings("disabled_modules", moduleStatus.disabled)) +} + +// runsSingletonModules 判断当前角色是否承担主动调度和初始化职责。 +func runsSingletonModules(role string) bool { + return role == constants.WorkerRoleAll || role == constants.WorkerRoleLeader +} + +// initWorkerRuntime 初始化所有角色共享的外部依赖和轮询运行时对象。 +func initWorkerRuntime(ctx context.Context, cfg *config.Config, appLogger *zap.Logger) *workerRuntime { redisAddr := cfg.Redis.Address + ":" + strconv.Itoa(cfg.Redis.Port) redisClient := redis.NewClient(&redis.Options{ Addr: redisAddr, @@ -76,52 +194,26 @@ func main() { ReadTimeout: cfg.Redis.ReadTimeout, WriteTimeout: cfg.Redis.WriteTimeout, }) - defer func() { - if err := redisClient.Close(); err != nil { - appLogger.Error("关闭 Redis 客户端失败", zap.Error(err)) - } - }() - // 测试 Redis 连接 - ctx := context.Background() if err := redisClient.Ping(ctx).Err(); err != nil { appLogger.Fatal("连接 Redis 失败", zap.Error(err)) } appLogger.Info("Redis 已连接", zap.String("address", redisAddr)) - // 初始化 PostgreSQL 连接 db, err := database.InitPostgreSQL(&cfg.Database, appLogger) if err != nil { appLogger.Fatal("初始化 PostgreSQL 失败", zap.Error(err)) } - defer func() { - sqlDB, _ := db.DB() - if sqlDB != nil { - if err := sqlDB.Close(); err != nil { - appLogger.Error("关闭 PostgreSQL 连接失败", zap.Error(err)) - } - } - }() - // 初始化对象存储服务（可选） storageSvc := initStorage(cfg, appLogger) - - // 初始化 Gateway 客户端（可选，用于轮询任务） gatewayClient := initGateway(cfg, appLogger) - // 创建 Asynq 客户端（用于调度器提交任务） asynqClient := asynq.NewClient(asynq.RedisClientOpt{ Addr: redisAddr, Password: cfg.Redis.Password, DB: cfg.Redis.DB, }) - defer func() { - if err := asynqClient.Close(); err != nil { - appLogger.Error("关闭 Asynq 客户端失败", zap.Error(err)) - } - }() - // 创建 Worker 依赖 workerDeps := &bootstrap.WorkerDependencies{ DB: db, Redis: redisClient, @@ -131,16 +223,13 @@ func main() { GatewayClient: gatewayClient, } - // Bootstrap Worker 组件 workerResult, err := bootstrap.BootstrapWorker(workerDeps) if err != nil { appLogger.Fatal("Worker Bootstrap 失败", zap.Error(err)) } - // 创建 Asynq Worker 服务器 workerServer := queue.NewServer(redisClient, &cfg.Queue, appLogger) - // 初始化轮询系统新组件（Phase 5.7） pollingConfigStore := postgres.NewPollingConfigStore(db) pollingConfigMgr := polling.NewPollingConfigManager(pollingConfigStore, redisClient, appLogger) if err := pollingConfigMgr.Load(ctx); err != nil { @@ -149,55 +238,125 @@ func main() { pollingConfigMgr.Start(ctx) pollingQueueMgr := polling.NewPollingQueueManager(redisClient, constants.PollingShardCount, appLogger) - pollingIotCardStore := postgres.NewIotCardStore(db, redisClient) - pollingBase := task.NewPollingBase(redisClient, pollingQueueMgr, pollingConfigMgr, pollingIotCardStore, appLogger, cfg.Polling.VerboseLog) + pollingBase := task.NewPollingBase( + redisClient, + pollingQueueMgr, + pollingConfigMgr, + pollingIotCardStore, + appLogger, + cfg.Polling.VerboseLog, + ) - // 后台渐进式初始化（将全量卡数据写入分片 Sorted Set） - pollingInitializer := polling.NewPollingInitializer(pollingIotCardStore, redisClient, pollingConfigMgr, pollingQueueMgr, appLogger) + pollingDeviceSimBindingStore := postgres.NewDeviceSimBindingStore(db, redisClient) + pollingDeviceStore := postgres.NewDeviceStore(db, redisClient) + lifecycleSvc := polling.NewPollingLifecycleService( + pollingQueueMgr, + pollingConfigMgr, + pollingIotCardStore, + pollingDeviceSimBindingStore, + pollingDeviceStore, + appLogger, + ) + + return &workerRuntime{ + redisAddr: redisAddr, + redisClient: redisClient, + db: db, + storageSvc: storageSvc, + gatewayClient: gatewayClient, + asynqClient: asynqClient, + workerResult: workerResult, + workerServer: workerServer, + pollingConfigMgr: pollingConfigMgr, + pollingQueueMgr: pollingQueueMgr, + pollingIotCardStore: pollingIotCardStore, + pollingBase: pollingBase, + lifecycleSvc: lifecycleSvc, + } +} + +// close 在 Worker 退出时关闭共享客户端与数据库连接。 +func (r *workerRuntime) close(appLogger *zap.Logger) { + if r.asynqClient != nil { + if err := r.asynqClient.Close(); err != nil { + appLogger.Error("关闭 Asynq 客户端失败", zap.Error(err)) + } + } + if r.db != nil { + sqlDB, _ := r.db.DB() + if sqlDB != nil { + if err := sqlDB.Close(); err != nil { + appLogger.Error("关闭 PostgreSQL 连接失败", zap.Error(err)) + } + } + } + if r.redisClient != nil { + if err := r.redisClient.Close(); err != nil { + appLogger.Error("关闭 Redis 客户端失败", zap.Error(err)) + } + } +} + +// startPollingInitializer 为 leader/all 启动渐进式初始化器，并保留配置变更重启逻辑。 +func startPollingInitializer(ctx context.Context, runtime *workerRuntime, appLogger *zap.Logger) *polling.PollingInitializer { + pollingInitializer := polling.NewPollingInitializer( + runtime.pollingIotCardStore, + runtime.redisClient, + runtime.pollingConfigMgr, + runtime.pollingQueueMgr, + appLogger, + ) pollingInitializer.StartBackground(ctx) - // 订阅配置变更事件：配置从空变为非空时重启 Initializer，将卡重新入队 - pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) { + runtime.pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) { if !hadConfigs && hasConfigs { appLogger.Info("轮询配置从空变为非空，触发队列重新初始化") pollingInitializer.Restart(ctx) } }) - // 创建生命周期服务（Worker 进程用，功能更完整） - pollingDeviceSimBindingStore := postgres.NewDeviceSimBindingStore(db, redisClient) - pollingDeviceStore := postgres.NewDeviceStore(db, redisClient) - lifecycleSvc := polling.NewPollingLifecycleService(pollingQueueMgr, pollingConfigMgr, pollingIotCardStore, pollingDeviceSimBindingStore, pollingDeviceStore, appLogger) + return pollingInitializer +} - // 初始化调度器（激活/重置 Handler 通过构造函数注入，防止遗漏） - dataResetHandler := polling.NewDataResetHandler(workerResult.Services.ResetService, appLogger) +// startPollingScheduler 为 leader/all 启动轮询调度器。 +func startPollingScheduler( + ctx context.Context, + runtime *workerRuntime, + pollingInitializer *polling.PollingInitializer, + appLogger *zap.Logger, +) *polling.Scheduler { + dataResetHandler := polling.NewDataResetHandler(runtime.workerResult.Services.ResetService, appLogger) activationHandler := polling.NewPackageActivationHandler( - db, redisClient, asynqClient, - workerResult.Services.ActivationService, - workerResult.Services.StopResumeService, + runtime.db, + runtime.redisClient, + runtime.asynqClient, + runtime.workerResult.Services.ActivationService, + runtime.workerResult.Services.StopResumeService, appLogger, ) - scheduler := polling.NewScheduler(redisClient, asynqClient, pollingQueueMgr, pollingConfigMgr, appLogger, activationHandler, dataResetHandler) - scheduler.SetInitializer(pollingInitializer) - if err := scheduler.Start(ctx); err != nil { + pollingScheduler := polling.NewScheduler( + runtime.redisClient, + runtime.asynqClient, + runtime.pollingQueueMgr, + runtime.pollingConfigMgr, + appLogger, + activationHandler, + dataResetHandler, + ) + pollingScheduler.SetInitializer(pollingInitializer) + + if err := pollingScheduler.Start(ctx); err != nil { appLogger.Error("启动轮询调度器失败", zap.Error(err)) - } else { - appLogger.Info("轮询调度器已启动") + return nil } - // 类型断言：StopResumeService 实际是 *iotCardSvc.StopResumeService，实现了 EvaluateAndAct 接口 - stopResumeSvc, _ := workerResult.Services.StopResumeService.(iot_card_svc.StopResumeServiceInterface) - // 创建任务处理器并注册 - taskHandler := queue.NewHandler(db, redisClient, storageSvc, gatewayClient, lifecycleSvc, workerResult, asynqClient, appLogger, pollingBase, stopResumeSvc) - taskHandler.RegisterHandlers() + return pollingScheduler +} - appLogger.Info("Worker 服务器配置完成", - zap.Int("concurrency", cfg.Queue.Concurrency), - zap.Any("queues", cfg.Queue.Queues)) - - // 创建 Asynq Scheduler（定时任务调度器：订单超时、告警检查、数据清理） +// startAsynqScheduler 为 leader/all 创建并启动 Asynq Scheduler。 +func startAsynqScheduler(cfg *config.Config, redisAddr string, appLogger *zap.Logger) *asynq.Scheduler { asynqScheduler := asynq.NewScheduler( asynq.RedisClientOpt{ Addr: redisAddr, @@ -207,57 +366,82 @@ func main() { &asynq.SchedulerOpts{Location: time.Local}, ) - // 注册定时任务：订单超时检查（每分钟） - if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask(constants.TaskTypeOrderExpire, nil)); err != nil { - appLogger.Fatal("注册订单超时定时任务失败", zap.Error(err)) - } - // 注册定时任务：告警检查（每分钟） - if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask(constants.TaskTypeAlertCheck, nil)); err != nil { - appLogger.Fatal("注册告警检查定时任务失败", zap.Error(err)) - } - // 注册定时任务：数据清理（每天凌晨 2 点） - if _, err := asynqScheduler.Register("0 2 * * *", asynq.NewTask(constants.TaskTypeDataCleanup, nil)); err != nil { - appLogger.Fatal("注册数据清理定时任务失败", zap.Error(err)) - } - // 注册定时任务：每日流量落盘（每天凌晨 2 点，超时 5 分钟） - if _, err := asynqScheduler.Register("0 2 * * *", asynq.NewTask(constants.TaskTypeDailyTrafficFlush, nil, asynq.MaxRetry(3), asynq.Timeout(5*time.Minute))); err != nil { - appLogger.Fatal("注册每日流量落盘定时任务失败", zap.Error(err)) + if err := registerAsynqScheduleTasks(asynqScheduler); err != nil { + appLogger.Fatal("注册 Asynq 定时任务失败", zap.Error(err)) } - // 启动 Asynq Scheduler go func() { if err := asynqScheduler.Run(); err != nil { appLogger.Fatal("Asynq Scheduler 启动失败", zap.Error(err)) } }() - appLogger.Info("Asynq Scheduler 已启动（订单超时: @every 1m, 告警检查: @every 1m, 数据清理: 0 2 * * *）") - // 优雅关闭 - quit := make(chan os.Signal, 1) - signal.Notify(quit, os.Interrupt, syscall.SIGTERM) + appLogger.Info("Asynq Scheduler 已启动（订单超时: @every 1m, 告警检查: @every 1m, 数据清理: 0 2 * * *, 每日流量落盘: 0 2 * * *）") + return asynqScheduler +} - // 启动 Worker 服务器（阻塞运行） - go func() { - if err := workerServer.Run(taskHandler.GetMux()); err != nil { - appLogger.Fatal("Worker 服务器运行失败", zap.Error(err)) - } - }() +// registerAsynqScheduleTasks 注册 Worker 入口需要的全部定时任务。 +func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler) error { + if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask(constants.TaskTypeOrderExpire, nil)); err != nil { + return fmt.Errorf("注册订单超时定时任务失败: %w", err) + } + if _, err := asynqScheduler.Register("@every 1m", asynq.NewTask(constants.TaskTypeAlertCheck, nil)); err != nil { + return fmt.Errorf("注册告警检查定时任务失败: %w", err) + } + if _, err := asynqScheduler.Register("0 2 * * *", asynq.NewTask(constants.TaskTypeDataCleanup, nil)); 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)), + ); err != nil { + return fmt.Errorf("注册每日流量落盘定时任务失败: %w", err) + } + return nil +} - appLogger.Info("Worker 服务器已启动") +// createTaskHandler 创建并返回包含全部任务处理器的 Asynq Handler。 +func createTaskHandler(runtime *workerRuntime, appLogger *zap.Logger) *queue.Handler { + stopResumeSvc, _ := runtime.workerResult.Services.StopResumeService.(iot_card_svc.StopResumeServiceInterface) + return queue.NewHandler( + runtime.db, + runtime.redisClient, + runtime.storageSvc, + runtime.gatewayClient, + runtime.lifecycleSvc, + runtime.workerResult, + runtime.asynqClient, + appLogger, + runtime.pollingBase, + stopResumeSvc, + ) +} - // 等待关闭信号 - <-quit +// shutdownWorker 按当前实例实际启动过的模块执行优雅关闭。 +func shutdownWorker( + cancel context.CancelFunc, + appLogger *zap.Logger, + workerServer *queue.Server, + pollingInitializer *polling.PollingInitializer, + pollingScheduler *polling.Scheduler, + asynqScheduler *asynq.Scheduler, +) { appLogger.Info("正在关闭 Worker 服务器...") - // 停止 Asynq Scheduler - asynqScheduler.Shutdown() + if asynqScheduler != nil { + asynqScheduler.Shutdown() + } + if pollingScheduler != nil { + pollingScheduler.Stop() + } - // 停止轮询调度器 - scheduler.Stop() + cancel() + + if pollingInitializer != nil { + pollingInitializer.Stop() + } - // 优雅关闭 Worker 服务器（等待正在执行的任务完成） workerServer.Shutdown() - appLogger.Info("Worker 服务器已停止") } diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 5a9ec01..d8d903d 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -138,6 +138,33 @@ | `JUNHONG_QUEUE_RETRY_MAX` | `3` | 最大重试次数 | | `JUNHONG_QUEUE_TIMEOUT` | `30m` | 任务超时时间 | +### Worker 运行配置 + +| 环境变量 | 默认值 | 说明 | +|---------|--------|------| +| `JUNHONG_WORKER_ROLE` | `all` | Worker 运行角色。`all` 为单实例兼容模式，`leader` 负责主动调度和初始化，`consumer` 只消费队列任务 | +| `JUNHONG_WORKER_INSTANCE_NAME` | `""` | Worker 实例名称，用于多实例日志区分，建议在多实例部署时配置唯一值 | + +**使用说明**： +- 单实例部署可不设置 `JUNHONG_WORKER_ROLE`，默认使用 `all`，行为与历史版本一致 +- 多实例部署推荐固定 `1 leader + N consumer` +- `leader` / `all` 只能保留一个实例承担主动调度职责，横向扩容时请新增 `consumer` + +**示例**： + +```bash +# 单实例兼容模式（默认可省略） +JUNHONG_WORKER_ROLE=all + +# 多实例 leader +JUNHONG_WORKER_ROLE=leader +JUNHONG_WORKER_INSTANCE_NAME=worker-leader-1 + +# 多实例 consumer +JUNHONG_WORKER_ROLE=consumer +JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-1 +``` + ### 限流中间件 | 环境变量 | 默认值 | 说明 | @@ -220,6 +247,8 @@ services: - JUNHONG_DATABASE_DBNAME=junhong_cmp - JUNHONG_REDIS_ADDRESS=redis - JUNHONG_JWT_SECRET_KEY=your-production-secret-key + - JUNHONG_WORKER_ROLE=all + - JUNHONG_WORKER_INSTANCE_NAME=worker-all-1 volumes: - ./logs:/app/logs @@ -234,6 +263,51 @@ services: image: redis:6 ``` +- 单实例部署默认就是 `all`，上例显式写出只是为了便于排查日志；不设置 `JUNHONG_WORKER_ROLE` 时效果相同 +- 多实例部署推荐 `1 leader + N consumer`，不要直接复制多个完整 `all` Worker + +```yaml +services: + worker-leader: + image: registry.boss160.cn/junhong/cmp-fiber-worker:latest + environment: + - JUNHONG_DATABASE_HOST=postgres + - JUNHONG_DATABASE_PORT=5432 + - JUNHONG_DATABASE_USER=junhong + - JUNHONG_DATABASE_PASSWORD=secret123 + - JUNHONG_DATABASE_DBNAME=junhong_cmp + - JUNHONG_REDIS_ADDRESS=redis + - JUNHONG_JWT_SECRET_KEY=your-production-secret-key + - JUNHONG_WORKER_ROLE=leader + - JUNHONG_WORKER_INSTANCE_NAME=worker-leader-1 + + worker-consumer-1: + image: registry.boss160.cn/junhong/cmp-fiber-worker:latest + environment: + - JUNHONG_DATABASE_HOST=postgres + - JUNHONG_DATABASE_PORT=5432 + - JUNHONG_DATABASE_USER=junhong + - JUNHONG_DATABASE_PASSWORD=secret123 + - JUNHONG_DATABASE_DBNAME=junhong_cmp + - JUNHONG_REDIS_ADDRESS=redis + - JUNHONG_JWT_SECRET_KEY=your-production-secret-key + - JUNHONG_WORKER_ROLE=consumer + - JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-1 + + worker-consumer-2: + image: registry.boss160.cn/junhong/cmp-fiber-worker:latest + environment: + - JUNHONG_DATABASE_HOST=postgres + - JUNHONG_DATABASE_PORT=5432 + - JUNHONG_DATABASE_USER=junhong + - JUNHONG_DATABASE_PASSWORD=secret123 + - JUNHONG_DATABASE_DBNAME=junhong_cmp + - JUNHONG_REDIS_ADDRESS=redis + - JUNHONG_JWT_SECRET_KEY=your-production-secret-key + - JUNHONG_WORKER_ROLE=consumer + - JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-2 +``` + ## 本地开发 本地开发可以创建 `.env` 文件（不要提交到 Git）： diff --git a/docs/polling-system/performance-tuning.md b/docs/polling-system/performance-tuning.md index f4e1ead..083a933 100644 --- a/docs/polling-system/performance-tuning.md +++ b/docs/polling-system/performance-tuning.md @@ -44,26 +44,35 @@ INSERT INTO tb_polling_concurrency_config (task_type, max_concurrency, descripti ### 3. Worker 多实例部署 -部署多个 Worker 实例分担负载： +多实例扩容时，必须拆分 Worker 角色，禁止直接复制多个完整 Worker 进程。推荐保留 `1 leader + N consumer`： ```yaml # docker-compose.yml 示例 services: - worker-1: + worker-leader: image: junhong-cmp-worker environment: - - WORKER_ID=1 - worker-2: + - JUNHONG_WORKER_ROLE=leader + - JUNHONG_WORKER_INSTANCE_NAME=worker-leader-1 + worker-consumer-1: image: junhong-cmp-worker environment: - - WORKER_ID=2 - worker-3: + - JUNHONG_WORKER_ROLE=consumer + - JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-1 + worker-consumer-2: image: junhong-cmp-worker environment: - - WORKER_ID=3 + - JUNHONG_WORKER_ROLE=consumer + - JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-2 ``` -注意：只需一个实例运行调度器，其他实例只处理任务。 +说明： +- `leader` 负责轮询初始化、轮询调度、Asynq 定时任务调度等单例职责 +- `consumer` 只负责 Asynq 队列消费，适合横向扩容 +- 单实例部署默认使用 `all`，行为等价于历史版本 +- 不要直接启动多个 `all` 或多个 `leader`，否则会重复扫库、重复调度、重复入队 +- 需要提升吞吐时，优先增加 `consumer` 数量，而不是复制完整 Worker +- 本期仅完成角色拆分，不包含 Redis Leader 锁、自动选主、任务唯一性去重，这些作为第二期增强处理 ### 4. 检查间隔优化 diff --git a/openspec/changes/archive/2026-05-06-split-worker-roles/.openspec.yaml b/openspec/changes/archive/2026-05-06-split-worker-roles/.openspec.yaml new file mode 100644 index 0000000..2188dbd --- /dev/null +++ b/openspec/changes/archive/2026-05-06-split-worker-roles/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-05-06 diff --git a/openspec/changes/archive/2026-05-06-split-worker-roles/design.md b/openspec/changes/archive/2026-05-06-split-worker-roles/design.md new file mode 100644 index 0000000..b786b7f --- /dev/null +++ b/openspec/changes/archive/2026-05-06-split-worker-roles/design.md @@ -0,0 +1,99 @@ +## Context + +当前 `cmd/worker/main.go` 的启动入口把两类职责放在同一进程中： + +- 可横向扩容的被动消费职责：Asynq Worker Server 和全部任务 Handler +- 必须单例运行的主动发起职责：`PollingInitializer`、`PollingScheduler`、套餐激活/流量重置扫描、Asynq Scheduler + +其中 Asynq Worker Server 天然适合多实例消费任务，但完整复制当前 worker 会同时复制主动调度模块。当前 Asynq v0.25.1 的 `Scheduler` 是进程内 cron，到点直接 `Enqueue`，不是跨进程单例；轮询初始化器也会启动后全量扫库并写 Redis 分片队列；流量重置节流状态是进程内字段。因此第一期必须先通过显式角色把部署语义收敛清楚。 + +本变更只处理“角色拆分和启动编排”，不改变已有 Handler、Service、Store、Model 的业务逻辑，不新增数据库表，也不改 Asynq 任务载荷。 + +## Goals / Non-Goals + +**Goals:** + +- 新增 `all`、`leader`、`consumer` 三种 Worker 角色。 +- 默认 `all`，旧部署不配置新环境变量时行为完全不变。 +- `consumer` 只运行任务消费相关模块，不运行轮询初始化器、轮询调度器和 Asynq Scheduler。 +- `leader` 与 `all` 的运行模块一致，但部署语义明确为单例主动调度实例。 +- 启动日志可直接证明当前实例的角色和模块启停情况。 +- 关闭流程按实际启动模块执行，避免 `consumer` 因未创建调度器而空指针或重复关闭。 + +**Non-Goals:** + +- 不实现 Redis Leader 锁或自动选主。 +- 不为套餐激活、订单超时等任务补充唯一性去重。 +- 不改 `PollingScheduler` 的调度算法和 Redis 分片队列实现。 +- 不调整 `queue.concurrency`、队列权重或业务 Handler 幂等逻辑。 +- 不新增 API、数据库迁移、DTO 或 OpenAPI 文档生成器改造。 +- 不新增自动化测试文件；按项目根规范使用编译和人工日志验证。 + +## Decisions + +### Decision 1：用显式配置角色，而不是自动探测或自动选主 + +**选择**：新增 `worker.role`，支持 `all`、`leader`、`consumer`，通过环境变量 `JUNHONG_WORKER_ROLE` 覆盖。 + +**备选方案**：启动时所有 worker 抢 Redis 锁，抢到锁者自动成为 leader。 + +**理由**：第一期目标是最小、安全、可回滚。显式角色能立即阻止完整进程被误横向复制，改动集中在配置和启动编排；自动选主需要续期、失锁处理、调度器停止/重启语义和故障切换策略，适合第二期单独设计。 + +### Decision 2：保留 `all` 作为默认兼容模式 + +**选择**：默认 `worker.role=all`。 + +**备选方案**：默认改为 `leader`。 + +**理由**：`all` 与当前运行行为完全一致，现有生产、本地开发和临时脚本不需要增加环境变量。`leader` 用于部署语义表达，等生产灰度确认后再从 `all` 切换到 `leader`。 + +### Decision 3：`consumer` 初始化共享依赖，但不启动单例模块 + +**选择**：`consumer` 仍初始化 Redis、PostgreSQL、Storage、Gateway、`BootstrapWorker`、`PollingConfigManager`、`PollingQueueManager`、`PollingBase`、`PollingLifecycleService`、TaskHandler 和 Asynq Worker Server。 + +**理由**：轮询任务 Handler 执行后需要读取轮询配置、执行并发控制、更新卡缓存、重入队分片队列；导入任务也可能触发轮询生命周期回调。因此 `consumer` 不是“瘦到只连 Asynq”的进程，而是“只不主动调度”的完整任务消费进程。 + +**边界**：`consumer` 不创建或不启动 `PollingInitializer`、`PollingScheduler`、Asynq Scheduler；如需为构造依赖创建可选对象，也必须保证不会启动其后台 goroutine。 + +### Decision 4：角色常量集中定义，启动逻辑用 helper 降低误用 + +**选择**：新增 Worker 角色常量，并提供类似 `RunsSingletonModules()` 的判断逻辑，避免在 `main.go` 多处硬编码字符串。 + +**位置约束**：遵循项目常量规范，角色字符串应放在 `pkg/constants/`，并添加中文注释；`pkg/config` 和 `cmd/worker` 通过常量判断角色。 + +**理由**：角色字符串会出现在配置校验、启动模块判断、日志输出和文档中，集中定义可以避免 `"consumer"` 这类 magic string 分散。 + +### Decision 5：关闭流程记录已启动模块，而不是假设全部存在 + +**选择**：为 Asynq Scheduler、Polling Scheduler、Polling Initializer 使用可选变量或启动状态标记；关闭时只关闭已启动模块。 + +**理由**：`consumer` 不启动单例模块，关闭逻辑如果无条件 `Shutdown/Stop` 会产生 nil panic 或关闭未运行对象。按启动状态关闭也让 `all/leader/consumer` 三种角色的生命周期可读。 + +### Decision 6：验证以编译和人工运行日志为准 + +**选择**：使用 `go build ./cmd/worker`、`go build ./...`、三种角色启动日志检查作为验收。 + +**理由**：项目根规范明确禁止自动化测试规划和 `*_test.go` 文件；本变更也主要是进程编排，人工日志能直接验证模块启停契约。 + +## Risks / Trade-offs + +| 风险 | 缓解措施 | +|------|----------| +| 部署层误把两个实例都配置成 `leader` | 第一期通过日志暴露误配；第二期再补 Redis Leader 锁 | +| `consumer` 漏初始化轮询依赖导致轮询任务无法重入队 | `consumer` 必须保留 `PollingConfigManager`、`PollingQueueManager`、`PollingBase`、`PollingLifecycleService` | +| 关闭流程引用未启动模块 | 使用可选变量/启动状态判断，只关闭已启动模块 | +| 默认值变更影响现有部署 | 默认保持 `all`，不配置环境变量时行为不变 | +| Asynq Scheduler 多实例去重被误解 | 文档和启动日志明确只有 `all/leader` 启动 Scheduler；`consumer` 禁止启动 | +| 角色字符串散落导致后续维护出错 | 角色值集中为常量，并在配置校验中拒绝非法值 | + +## Migration Plan + +1. 发布代码后，不修改现有环境变量，单实例继续以默认 `all` 运行。 +2. 灰度将唯一 worker 设置为 `JUNHONG_WORKER_ROLE=leader`，确认日志中初始化器、轮询调度器和 Asynq Scheduler 均启动。 +3. 新增第一个 `consumer` 实例，设置 `JUNHONG_WORKER_ROLE=consumer` 和唯一 `JUNHONG_WORKER_INSTANCE_NAME`。 +4. 检查 `consumer` 日志，确认未出现“轮询调度器已启动”和“Asynq Scheduler 已启动”，且 Asynq Worker Server 正常启动。 +5. 如发生异常，先停止所有 `consumer`，保留单个 `leader`；若仍异常，将唯一实例切回默认 `all` 或回滚二进制。 + +## Open Questions + +无。自动选主、Leader 分布式锁和任务唯一性去重作为第二期增强，不阻塞第一期实施。 diff --git a/openspec/changes/archive/2026-05-06-split-worker-roles/proposal.md b/openspec/changes/archive/2026-05-06-split-worker-roles/proposal.md new file mode 100644 index 0000000..73a8fba --- /dev/null +++ b/openspec/changes/archive/2026-05-06-split-worker-roles/proposal.md @@ -0,0 +1,63 @@ +## Why + +当前 `cmd/worker` 同时承担 Asynq 任务消费、轮询初始化、轮询调度、套餐激活/流量重置扫描和 Asynq 定时任务调度职责，导致完整 worker 进程不能安全横向复制。为了支持后续多实例扩容，同时避免重复扫库、重复调度和重复入队，本变更先引入显式 Worker 角色，将“可多开的消费职责”和“必须单例的主动调度职责”拆开。 + +## What Changes + +- 新增 Worker 运行角色配置： + - `all`：兼容当前单实例行为，启动全部模块 + - `leader`：明确承担单例职责，启动全部模块 + - `consumer`：只启动 Asynq Worker Server 和任务处理依赖，不启动任何主动调度/初始化模块 +- 新增实例名称配置 `JUNHONG_WORKER_INSTANCE_NAME`，用于日志区分多实例。 +- 修改 `cmd/worker/main.go` 启动流程，将共享依赖初始化和按角色启停模块分离。 +- `consumer` 仍注册全部任务 Handler，确保可以消费轮询任务、套餐激活任务、订单超时任务、告警任务、数据清理任务等已有 Asynq 任务。 +- 启动日志明确打印当前角色、实例名、启用模块和禁用模块，便于部署验证和排障。 +- 保持默认 `all`，不修改现有部署配置时行为完全不变。 +- 不引入自动选主、Redis Leader 锁或任务唯一性去重；这些作为第二期增强，不纳入本次最小改造。 + +## Capabilities + +### New Capabilities + +- `worker-role-management`：定义 Worker 运行角色、按角色启停模块、兼容模式、消费者扩容语义和启动日志要求。 + +### Modified Capabilities + +- `embedded-config`：新增 `worker.role`、`worker.instance_name` 默认配置和环境变量覆盖要求。 + +## Impact + +**功能 ID**：`feature-worker-role-split` + +**涉及文件**： +- `cmd/worker/main.go`：按角色控制 `PollingInitializer`、`PollingScheduler`、`Asynq Scheduler` 启停 +- `pkg/config/config.go`：新增 Worker 配置结构和角色校验 +- `pkg/config/defaults/config.yaml`：新增默认 worker 配置节 +- `pkg/config/loader.go`：绑定 `worker.role`、`worker.instance_name` +- `docs/environment-variables.md`：补充 Worker 角色环境变量和部署说明 + +**技术栈合规性**： +- 配置继续使用 Viper 加载和环境变量覆盖 +- 日志继续使用 Zap 输出中文日志 +- 异步任务继续使用 Asynq，不新增队列依赖 +- 不涉及 Fiber Handler、GORM Store、数据库迁移或 DTO 变更 + +**架构分层**： +- 本变更属于进程编排和配置层改造，不新增业务 Handler/Service/Store/Model +- 已有任务处理链路保持不变，仍由 Asynq Worker Server 分发到 `pkg/queue` 注册的 Handler + +**验证计划**： +- 按项目根规范，本变更不新增自动化测试或 `*_test.go` 文件 +- 使用 `go build ./cmd/worker` 验证 worker 入口编译通过 +- 使用 `go build ./...` 验证项目整体编译通过 +- 人工以 `all`、`leader`、`consumer` 三种配置启动/观察日志，确认模块启停符合角色语义 +- 人工检查 `consumer` 日志不出现“轮询调度器已启动”和“Asynq Scheduler 已启动” + +**性能影响**： +- 单实例默认 `all` 行为不变，无额外运行时开销 +- `consumer` 不运行初始化器和调度器，可降低多实例场景下重复 DB/Redis 压力 +- 多实例阶段应优先扩展 `consumer`，不要复制 `leader` + +**兼容性与回滚**： +- 默认值为 `all`，旧部署无需新增环境变量即可保持当前行为 +- 如上线后有问题，可停止所有 `consumer`，保留单个 `leader`；必要时将唯一实例切回 `all` diff --git a/openspec/changes/archive/2026-05-06-split-worker-roles/specs/embedded-config/spec.md b/openspec/changes/archive/2026-05-06-split-worker-roles/specs/embedded-config/spec.md new file mode 100644 index 0000000..739857b --- /dev/null +++ b/openspec/changes/archive/2026-05-06-split-worker-roles/specs/embedded-config/spec.md @@ -0,0 +1,45 @@ +## MODIFIED Requirements + +### Requirement: 配置嵌入 + +系统 SHALL 使用 Go 的 `go:embed` 指令将默认配置文件嵌入二进制文件。 + +嵌入文件位置：`pkg/config/defaults/config.yaml` + +#### Scenario: 加载嵌入配置 + +- **WHEN** 调用 `config.Load()` +- **THEN** 系统从嵌入的 `defaults/config.yaml` 读取默认配置 +- **AND** 无需外部配置文件即可启动 + +#### Scenario: 嵌入配置包含完整结构 + +- **WHEN** 读取嵌入配置 +- **THEN** 配置包含所有配置节：server、database、redis、storage、logging、queue、jwt、middleware、worker + +## ADDED Requirements + +### Requirement: Worker 运行配置 + +系统 SHALL 在嵌入默认配置中提供 `worker` 配置节，并支持通过 `JUNHONG_WORKER_ROLE` 和 `JUNHONG_WORKER_INSTANCE_NAME` 环境变量覆盖。 + +默认配置： +- `worker.role`: `all` +- `worker.instance_name`: 空字符串 + +#### Scenario: 默认 Worker 角色 +- **WHEN** 未设置 `JUNHONG_WORKER_ROLE` +- **THEN** `config.Load()` 返回的 `Config.Worker.Role` 为 `all` + +#### Scenario: 环境变量覆盖 Worker 角色 +- **WHEN** 设置环境变量 `JUNHONG_WORKER_ROLE=consumer` +- **THEN** `config.Load()` 返回的 `Config.Worker.Role` 为 `consumer` + +#### Scenario: 环境变量覆盖实例名称 +- **WHEN** 设置环境变量 `JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-1` +- **THEN** `config.Load()` 返回的 `Config.Worker.InstanceName` 为 `worker-consumer-1` + +#### Scenario: Worker 角色参与配置校验 +- **WHEN** 设置环境变量 `JUNHONG_WORKER_ROLE=invalid` +- **THEN** `config.Load()` 返回错误 +- **AND** 错误信息明确指出 `worker.role` 只允许 `all`、`leader`、`consumer` diff --git a/openspec/changes/archive/2026-05-06-split-worker-roles/specs/worker-role-management/spec.md b/openspec/changes/archive/2026-05-06-split-worker-roles/specs/worker-role-management/spec.md new file mode 100644 index 0000000..6ab07d7 --- /dev/null +++ b/openspec/changes/archive/2026-05-06-split-worker-roles/specs/worker-role-management/spec.md @@ -0,0 +1,125 @@ +## ADDED Requirements + +### Requirement: Worker 角色配置 + +系统 SHALL 支持 `all`、`leader`、`consumer` 三种 Worker 运行角色，并在启动时根据配置决定当前进程职责。角色值 MUST 使用小写英文字符串，非法角色 MUST 使配置加载失败并返回明确错误。 + +#### Scenario: 默认兼容模式 +- **WHEN** 未设置 `JUNHONG_WORKER_ROLE` +- **THEN** Worker 使用默认角色 `all` +- **AND** 启动行为与变更前单实例 worker 保持一致 + +#### Scenario: 非法角色启动失败 +- **WHEN** 设置 `JUNHONG_WORKER_ROLE=scheduler` +- **THEN** `config.Load()` 返回配置校验错误 +- **AND** Worker 不进入 Redis、PostgreSQL 或 Asynq 启动流程 + +#### Scenario: 实例名称可为空 +- **WHEN** 未设置 `JUNHONG_WORKER_INSTANCE_NAME` +- **THEN** Worker 可以正常启动 +- **AND** 启动日志中的实例名称为空或显示默认占位值 + +### Requirement: 按角色启停 Worker 模块 + +系统 SHALL 将 Worker 启动流程拆分为共享依赖初始化和角色模块启停两部分。所有角色 MUST 启动 Asynq Worker Server；只有 `all` 和 `leader` SHALL 启动主动调度与初始化模块。 + +**共享模块**： +- Redis 客户端 +- PostgreSQL 连接 +- Storage 服务（可选） +- Gateway 客户端（可选） +- Asynq Client +- `BootstrapWorker` +- Asynq Worker Server +- `PollingConfigManager` +- `PollingQueueManager` +- `PollingBase` +- `PollingLifecycleService` +- TaskHandler 及全部任务 Handler 注册 + +**单例模块**： +- `PollingInitializer` +- `PollingScheduler` +- Asynq Scheduler + +#### Scenario: all 角色启动全部模块 +- **WHEN** `worker.role=all` +- **THEN** Worker 启动 Asynq Worker Server +- **AND** 启动 `PollingInitializer` +- **AND** 启动 `PollingScheduler` +- **AND** 启动 Asynq Scheduler + +#### Scenario: leader 角色启动全部模块 +- **WHEN** `worker.role=leader` +- **THEN** Worker 启动 Asynq Worker Server +- **AND** 启动 `PollingInitializer` +- **AND** 启动 `PollingScheduler` +- **AND** 启动 Asynq Scheduler +- **AND** 日志表达该实例承担单例主动调度职责 + +#### Scenario: consumer 角色只启动任务消费模块 +- **WHEN** `worker.role=consumer` +- **THEN** Worker 启动 Asynq Worker Server +- **AND** 注册全部任务 Handler +- **AND** 不启动 `PollingInitializer` +- **AND** 不启动 `PollingScheduler` +- **AND** 不启动 Asynq Scheduler + +### Requirement: consumer 保留轮询任务运行依赖 + +`consumer` 角色虽然不主动调度，但系统 SHALL 保留轮询任务 Handler 所需依赖，使其可以正常消费 Leader 投递的轮询任务并在任务完成后重新入队。 + +#### Scenario: consumer 消费轮询任务后可重新入队 +- **WHEN** `consumer` 实例消费 `TaskTypePollingCarddata` 任务 +- **THEN** Handler 可通过 `PollingBase` 读取轮询配置和 Redis 并发控制 +- **AND** Handler 可通过 `PollingQueueManager` 将卡重新写回分片 Sorted Set + +#### Scenario: consumer 可处理导入任务的轮询生命周期回调 +- **WHEN** `consumer` 实例消费 IoT 卡导入任务并创建新卡 +- **THEN** 导入任务可通过 `PollingLifecycleService` 将新卡按配置加入轮询分片队列 +- **AND** 该操作不依赖本实例启动 `PollingScheduler` + +### Requirement: Worker 启动日志 + +系统 SHALL 在 Worker 启动时输出当前角色、实例名称、启用模块和禁用模块，日志消息 MUST 使用中文，便于部署后通过日志确认实际运行职责。 + +#### Scenario: consumer 日志显示禁用单例模块 +- **WHEN** `worker.role=consumer` +- **THEN** 启动日志包含 `Worker 角色` +- **AND** 启动日志包含 `实例名称` +- **AND** 启动日志显示已禁用 `polling_initializer`、`polling_scheduler`、`asynq_scheduler` + +#### Scenario: leader 日志显示启用单例模块 +- **WHEN** `worker.role=leader` +- **THEN** 启动日志显示已启用 `queue_server`、`polling_initializer`、`polling_scheduler`、`asynq_scheduler` + +### Requirement: 按启动状态优雅关闭 + +系统 SHALL 只关闭当前角色实际启动过的后台模块，避免 `consumer` 对未启动模块执行 `Stop` 或 `Shutdown`。 + +#### Scenario: consumer 关闭不访问未启动调度器 +- **WHEN** `consumer` 收到退出信号 +- **THEN** Worker 关闭 Asynq Worker Server +- **AND** 不调用未启动的 `PollingScheduler.Stop` +- **AND** 不调用未启动的 Asynq Scheduler `Shutdown` + +#### Scenario: leader 关闭全部已启动模块 +- **WHEN** `leader` 收到退出信号 +- **THEN** Worker 关闭 Asynq Scheduler +- **AND** 停止 `PollingScheduler` +- **AND** 优雅关闭 Asynq Worker Server + +### Requirement: 多实例部署语义 + +系统 SHALL 支持“1 个 `leader` + N 个 `consumer`”的部署形态。部署时 SHOULD 固定只有一个 `leader` 或 `all` 实例承担主动调度职责，横向扩容 SHALL 优先增加 `consumer`。 + +#### Scenario: 新增 consumer 不复制主动调度 +- **WHEN** 生产已有一个 `leader`，再新增一个 `consumer` +- **THEN** 定时任务仍只由 `leader` 主动提交 +- **AND** 轮询全量初始化仍只由 `leader` 执行 +- **AND** `consumer` 只从 Asynq 队列消费已有任务 + +#### Scenario: 回滚到单实例兼容模式 +- **WHEN** 多实例部署出现问题并停止所有 `consumer` +- **THEN** 保留一个 `leader` 可继续运行完整 Worker 功能 +- **AND** 必要时可将唯一实例配置切回 `all` 恢复兼容模式 diff --git a/openspec/changes/archive/2026-05-06-split-worker-roles/tasks.md b/openspec/changes/archive/2026-05-06-split-worker-roles/tasks.md new file mode 100644 index 0000000..e62b751 --- /dev/null +++ b/openspec/changes/archive/2026-05-06-split-worker-roles/tasks.md @@ -0,0 +1,38 @@ +## 1. 配置与常量 + +- [x] 1.1 在 `pkg/constants/` 新增 Worker 角色常量：`all`、`leader`、`consumer`，并为所有导出常量添加中文注释 +- [x] 1.2 在 `pkg/config/config.go` 新增 `WorkerConfig`，字段包含 `Role` 和 `InstanceName` +- [x] 1.3 在 `Config` 中挂载 `Worker WorkerConfig`，并在 `Validate()` 中校验 `worker.role` 只允许 `all`、`leader`、`consumer` +- [x] 1.4 在 `pkg/config/defaults/config.yaml` 新增 `worker.role: "all"` 和 `worker.instance_name: ""` +- [x] 1.5 在 `pkg/config/loader.go` 绑定 `worker.role` 和 `worker.instance_name` 环境变量 +- [x] 1.6 运行 `lsp_diagnostics` 检查 `pkg/config/config.go`、`pkg/config/loader.go`、新增常量文件无诊断错误 + +## 2. Worker 启动角色拆分 + +- [x] 2.1 在 `cmd/worker/main.go` 中读取 `cfg.Worker.Role` 和 `cfg.Worker.InstanceName`，并输出当前角色和实例名称日志 +- [x] 2.2 将当前 Worker 启动流程整理为共享依赖初始化段，确保 Redis、PostgreSQL、Storage、Gateway、Asynq Client、`BootstrapWorker`、`PollingConfigManager`、`PollingQueueManager`、`PollingBase`、`PollingLifecycleService` 对所有角色都可用 +- [x] 2.3 为 `all/leader` 启动 `PollingInitializer`，并保留配置从空变为非空时触发 `Restart(ctx)` 的 WatchChanges 逻辑 +- [x] 2.4 为 `all/leader` 创建并启动 `PollingScheduler`，为 `consumer` 跳过调度器创建或启动 +- [x] 2.5 为 `all/leader` 创建、注册并启动 Asynq Scheduler，注册现有 `order_expire`、`alert_check`、`data_cleanup`、`daily_traffic_flush` 定时任务；为 `consumer` 完全跳过 +- [x] 2.6 确保所有角色都创建 Asynq Worker Server、注册全部任务 Handler，并调用 `workerServer.Run(taskHandler.GetMux())` +- [x] 2.7 启动日志输出已启用模块和已禁用模块，模块名包含 `queue_server`、`polling_initializer`、`polling_scheduler`、`asynq_scheduler` +- [x] 2.8 调整优雅关闭逻辑：只关闭当前角色实际启动过的 Asynq Scheduler 和 Polling Scheduler，所有角色都关闭 Worker Server +- [x] 2.9 运行 `lsp_diagnostics` 检查 `cmd/worker/main.go` 无诊断错误 + +## 3. 文档更新 + +- [x] 3.1 更新 `docs/environment-variables.md`，新增 `JUNHONG_WORKER_ROLE` 和 `JUNHONG_WORKER_INSTANCE_NAME` 说明、默认值和示例 +- [x] 3.2 在 `docs/environment-variables.md` 的 Docker Compose 示例中说明单实例默认 `all`，多实例推荐 `1 leader + N consumer` +- [x] 3.3 更新 `docs/polling-system/performance-tuning.md` 的 Worker 多实例部署段落，明确禁止直接复制完整 worker，推荐扩展 `consumer` +- [x] 3.4 对更新过的文档执行人工阅读检查，确认无“多个完整 worker 可直接多开”的歧义表述 + +## 4. 编译与人工验证 + +- [x] 4.1 对修改过的 Go 文件执行 `gofmt` +- [x] 4.2 执行 `go build ./cmd/worker`，确认 Worker 入口编译通过 +- [x] 4.3 执行 `go build ./...`，确认项目整体编译通过 +- [x] 4.4 使用默认配置值检查 `all` 模式日志预期：应显示启用 `queue_server`、`polling_initializer`、`polling_scheduler`、`asynq_scheduler` +- [x] 4.5 使用 `JUNHONG_WORKER_ROLE=leader` 检查日志预期：应显示与 `all` 相同的启用模块，并体现角色为 `leader` +- [x] 4.6 使用 `JUNHONG_WORKER_ROLE=consumer` 检查日志预期：应显示启用 `queue_server`，禁用 `polling_initializer`、`polling_scheduler`、`asynq_scheduler` +- [x] 4.7 使用非法角色值检查配置校验预期：`JUNHONG_WORKER_ROLE=invalid` 时启动失败，错误信息明确指向 `worker.role` +- [x] 4.8 记录未覆盖风险：本期不包含 Redis Leader 锁、自动选主、任务唯一性去重，这些进入第二期增强 diff --git a/openspec/specs/embedded-config/spec.md b/openspec/specs/embedded-config/spec.md index 2a8039f..1042161 100644 --- a/openspec/specs/embedded-config/spec.md +++ b/openspec/specs/embedded-config/spec.md @@ -18,7 +18,7 @@ TBD - created by archiving change deployment-self-init. Update Purpose after arc #### Scenario: 嵌入配置包含完整结构 - **WHEN** 读取嵌入配置 -- **THEN** 配置包含所有配置节：server、database、redis、storage、logging、queue、jwt、middleware +- **THEN** 配置包含所有配置节：server、database、redis、storage、logging、queue、jwt、middleware、worker ### Requirement: 环境变量覆盖 @@ -42,6 +42,31 @@ TBD - created by archiving change deployment-self-init. Update Purpose after arc - **WHEN** 未设置某个配置的环境变量 - **THEN** 使用嵌入配置中的默认值 +### Requirement: Worker 运行配置 + +系统 SHALL 在嵌入默认配置中提供 `worker` 配置节，并支持通过 `JUNHONG_WORKER_ROLE` 和 `JUNHONG_WORKER_INSTANCE_NAME` 环境变量覆盖。 + +默认配置： +- `worker.role`: `all` +- `worker.instance_name`: 空字符串 + +#### Scenario: 默认 Worker 角色 +- **WHEN** 未设置 `JUNHONG_WORKER_ROLE` +- **THEN** `config.Load()` 返回的 `Config.Worker.Role` 为 `all` + +#### Scenario: 环境变量覆盖 Worker 角色 +- **WHEN** 设置环境变量 `JUNHONG_WORKER_ROLE=consumer` +- **THEN** `config.Load()` 返回的 `Config.Worker.Role` 为 `consumer` + +#### Scenario: 环境变量覆盖实例名称 +- **WHEN** 设置环境变量 `JUNHONG_WORKER_INSTANCE_NAME=worker-consumer-1` +- **THEN** `config.Load()` 返回的 `Config.Worker.InstanceName` 为 `worker-consumer-1` + +#### Scenario: Worker 角色参与配置校验 +- **WHEN** 设置环境变量 `JUNHONG_WORKER_ROLE=invalid` +- **THEN** `config.Load()` 返回错误 +- **AND** 错误信息明确指出 `worker.role` 只允许 `all`、`leader`、`consumer` + ### Requirement: 配置优先级 系统 SHALL 按以下优先级应用配置（高到低）： @@ -130,4 +155,3 @@ TBD - created by archiving change deployment-self-init. Update Purpose after arc - **WHEN** 读取嵌入配置 - **THEN** 敏感配置项的值为空字符串 - **AND** 必须通过环境变量提供实际值 - diff --git a/openspec/specs/worker-role-management/spec.md b/openspec/specs/worker-role-management/spec.md new file mode 100644 index 0000000..de2c4e3 --- /dev/null +++ b/openspec/specs/worker-role-management/spec.md @@ -0,0 +1,130 @@ +# worker-role-management Specification + +## Purpose +TBD - created by archiving change split-worker-roles. Update Purpose after archive. + +## Requirements + +### Requirement: Worker 角色配置 + +系统 SHALL 支持 `all`、`leader`、`consumer` 三种 Worker 运行角色，并在启动时根据配置决定当前进程职责。角色值 MUST 使用小写英文字符串，非法角色 MUST 使配置加载失败并返回明确错误。 + +#### Scenario: 默认兼容模式 +- **WHEN** 未设置 `JUNHONG_WORKER_ROLE` +- **THEN** Worker 使用默认角色 `all` +- **AND** 启动行为与变更前单实例 worker 保持一致 + +#### Scenario: 非法角色启动失败 +- **WHEN** 设置 `JUNHONG_WORKER_ROLE=scheduler` +- **THEN** `config.Load()` 返回配置校验错误 +- **AND** Worker 不进入 Redis、PostgreSQL 或 Asynq 启动流程 + +#### Scenario: 实例名称可为空 +- **WHEN** 未设置 `JUNHONG_WORKER_INSTANCE_NAME` +- **THEN** Worker 可以正常启动 +- **AND** 启动日志中的实例名称为空或显示默认占位值 + +### Requirement: 按角色启停 Worker 模块 + +系统 SHALL 将 Worker 启动流程拆分为共享依赖初始化和角色模块启停两部分。所有角色 MUST 启动 Asynq Worker Server；只有 `all` 和 `leader` SHALL 启动主动调度与初始化模块。 + +**共享模块**： +- Redis 客户端 +- PostgreSQL 连接 +- Storage 服务（可选） +- Gateway 客户端（可选） +- Asynq Client +- `BootstrapWorker` +- Asynq Worker Server +- `PollingConfigManager` +- `PollingQueueManager` +- `PollingBase` +- `PollingLifecycleService` +- TaskHandler 及全部任务 Handler 注册 + +**单例模块**： +- `PollingInitializer` +- `PollingScheduler` +- Asynq Scheduler + +#### Scenario: all 角色启动全部模块 +- **WHEN** `worker.role=all` +- **THEN** Worker 启动 Asynq Worker Server +- **AND** 启动 `PollingInitializer` +- **AND** 启动 `PollingScheduler` +- **AND** 启动 Asynq Scheduler + +#### Scenario: leader 角色启动全部模块 +- **WHEN** `worker.role=leader` +- **THEN** Worker 启动 Asynq Worker Server +- **AND** 启动 `PollingInitializer` +- **AND** 启动 `PollingScheduler` +- **AND** 启动 Asynq Scheduler +- **AND** 日志表达该实例承担单例主动调度职责 + +#### Scenario: consumer 角色只启动任务消费模块 +- **WHEN** `worker.role=consumer` +- **THEN** Worker 启动 Asynq Worker Server +- **AND** 注册全部任务 Handler +- **AND** 不启动 `PollingInitializer` +- **AND** 不启动 `PollingScheduler` +- **AND** 不启动 Asynq Scheduler + +### Requirement: consumer 保留轮询任务运行依赖 + +`consumer` 角色虽然不主动调度，但系统 SHALL 保留轮询任务 Handler 所需依赖，使其可以正常消费 Leader 投递的轮询任务并在任务完成后重新入队。 + +#### Scenario: consumer 消费轮询任务后可重新入队 +- **WHEN** `consumer` 实例消费 `TaskTypePollingCarddata` 任务 +- **THEN** Handler 可通过 `PollingBase` 读取轮询配置和 Redis 并发控制 +- **AND** Handler 可通过 `PollingQueueManager` 将卡重新写回分片 Sorted Set + +#### Scenario: consumer 可处理导入任务的轮询生命周期回调 +- **WHEN** `consumer` 实例消费 IoT 卡导入任务并创建新卡 +- **THEN** 导入任务可通过 `PollingLifecycleService` 将新卡按配置加入轮询分片队列 +- **AND** 该操作不依赖本实例启动 `PollingScheduler` + +### Requirement: Worker 启动日志 + +系统 SHALL 在 Worker 启动时输出当前角色、实例名称、启用模块和禁用模块，日志消息 MUST 使用中文，便于部署后通过日志确认实际运行职责。 + +#### Scenario: consumer 日志显示禁用单例模块 +- **WHEN** `worker.role=consumer` +- **THEN** 启动日志包含 `Worker 角色` +- **AND** 启动日志包含 `实例名称` +- **AND** 启动日志显示已禁用 `polling_initializer`、`polling_scheduler`、`asynq_scheduler` + +#### Scenario: leader 日志显示启用单例模块 +- **WHEN** `worker.role=leader` +- **THEN** 启动日志显示已启用 `queue_server`、`polling_initializer`、`polling_scheduler`、`asynq_scheduler` + +### Requirement: 按启动状态优雅关闭 + +系统 SHALL 只关闭当前角色实际启动过的后台模块，避免 `consumer` 对未启动模块执行 `Stop` 或 `Shutdown`。 + +#### Scenario: consumer 关闭不访问未启动调度器 +- **WHEN** `consumer` 收到退出信号 +- **THEN** Worker 关闭 Asynq Worker Server +- **AND** 不调用未启动的 `PollingScheduler.Stop` +- **AND** 不调用未启动的 Asynq Scheduler `Shutdown` + +#### Scenario: leader 关闭全部已启动模块 +- **WHEN** `leader` 收到退出信号 +- **THEN** Worker 关闭 Asynq Scheduler +- **AND** 停止 `PollingScheduler` +- **AND** 优雅关闭 Asynq Worker Server + +### Requirement: 多实例部署语义 + +系统 SHALL 支持“1 个 `leader` + N 个 `consumer`”的部署形态。部署时 SHOULD 固定只有一个 `leader` 或 `all` 实例承担主动调度职责，横向扩容 SHALL 优先增加 `consumer`。 + +#### Scenario: 新增 consumer 不复制主动调度 +- **WHEN** 生产已有一个 `leader`，再新增一个 `consumer` +- **THEN** 定时任务仍只由 `leader` 主动提交 +- **AND** 轮询全量初始化仍只由 `leader` 执行 +- **AND** `consumer` 只从 Asynq 队列消费已有任务 + +#### Scenario: 回滚到单实例兼容模式 +- **WHEN** 多实例部署出现问题并停止所有 `consumer` +- **THEN** 保留一个 `leader` 可继续运行完整 Worker 功能 +- **AND** 必要时可将唯一实例配置切回 `all` 恢复兼容模式 diff --git a/pkg/config/config.go b/pkg/config/config.go index 22d5f9f..308b749 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -6,6 +6,8 @@ import ( "strings" "sync/atomic" "time" + + "github.com/break/junhong_cmp_fiber/pkg/constants" ) // globalConfig 保存当前配置，支持原子访问 @@ -27,6 +29,7 @@ type Config struct { Gateway GatewayConfig `mapstructure:"gateway"` PollingAutoTrigger PollingAutoTriggerConfig `mapstructure:"polling_auto_trigger"` Polling PollingConfig `mapstructure:"polling"` + Worker WorkerConfig `mapstructure:"worker"` } // ServerConfig HTTP 服务器配置 @@ -147,6 +150,12 @@ type GatewayConfig struct { Timeout int `mapstructure:"timeout"` // 超时时间（秒） } +// WorkerConfig Worker 进程运行配置 +type WorkerConfig struct { + Role string `mapstructure:"role"` // Worker 运行角色：all、leader、consumer + InstanceName string `mapstructure:"instance_name"` // Worker 实例名称，用于多实例日志区分 +} + // PollingConfig 轮询通用配置 type PollingConfig struct { // VerboseLog 开启后，所有轮询 handler 在成功获取上游数据时以 Info 级别输出详细日志 @@ -320,6 +329,21 @@ func (c *Config) Validate() error { } } + validWorkerRoles := map[string]bool{ + constants.WorkerRoleAll: true, + constants.WorkerRoleLeader: true, + constants.WorkerRoleConsumer: true, + } + if !validWorkerRoles[c.Worker.Role] { + return fmt.Errorf( + "invalid configuration: worker.role: invalid worker role (current value: %s, expected: %s, %s, %s)", + c.Worker.Role, + constants.WorkerRoleAll, + constants.WorkerRoleLeader, + constants.WorkerRoleConsumer, + ) + } + return nil } diff --git a/pkg/config/defaults/config.yaml b/pkg/config/defaults/config.yaml index 935d33a..5e5f102 100644 --- a/pkg/config/defaults/config.yaml +++ b/pkg/config/defaults/config.yaml @@ -125,3 +125,7 @@ polling_auto_trigger: enable_auto_trigger: true auto_trigger_system_user_id: 1 # 暂用 SuperAdmin，生产环境建议创建专用平台账号并更新此值 +# Worker 运行配置 +worker: + role: "all" + instance_name: "" diff --git a/pkg/config/loader.go b/pkg/config/loader.go index f0f23d5..ff05ee2 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -118,6 +118,8 @@ func bindEnvVariables(v *viper.Viper) { "gateway.app_id", "gateway.app_secret", "gateway.timeout", + "worker.role", + "worker.instance_name", "wechat.official_account.app_id", "wechat.official_account.app_secret", "wechat.official_account.token", diff --git a/pkg/constants/worker.go b/pkg/constants/worker.go new file mode 100644 index 0000000..618072e --- /dev/null +++ b/pkg/constants/worker.go @@ -0,0 +1,11 @@ +package constants + +// Worker 角色常量 +const ( + // WorkerRoleAll 兼容单实例模式，启动全部 Worker 模块 + WorkerRoleAll = "all" + // WorkerRoleLeader 单例调度模式，启动全部 Worker 模块 + WorkerRoleLeader = "leader" + // WorkerRoleConsumer 消费者模式，只启动队列消费相关模块 + WorkerRoleConsumer = "consumer" +) diff --git a/worker-多实例升级方案.md b/worker-多实例升级方案.md new file mode 100644 index 0000000..fdef7b3 --- /dev/null +++ b/worker-多实例升级方案.md @@ -0,0 +1,550 @@ +# Worker 多实例升级方案 + +## 1. 背景 + +当前 `cmd/worker/main.go` 不是一个纯粹的“任务消费进程”，而是把以下几类职责放在了同一个入口里： + +1. Asynq 任务消费器 +2. 轮询配置管理器 +3. 轮询全量初始化器 +4. 轮询调度器 +5. Asynq 定时任务调度器 + +这导致一个关键事实： + +- **Asynq 消费器天然适合多实例横向扩容** +- **调度器、初始化器这类“主动发起工作”的职责不适合直接整进程复制** + +因此，当前阶段不能简单把整个 `worker` 容器横向扩成多个副本，否则会出现重复调度、重复扫库、重复初始化、重复入队等问题。 + +本方案的目标不是否定多实例，而是把当前 `worker` 升级成： + +- **1 个单例 Leader Worker** +- **N 个可横向扩展的 Consumer Worker** + +这样既保留未来扩容能力，又能控制当前风险。 + +--- + +## 2. 当前现状 + +### 2.1 当前 `worker` 进程承担的职责 + +当前入口集中在 `cmd/worker/main.go:140-261`，关键行为如下： + +| 模块 | 位置 | 当前行为 | 是否适合多开 | +|------|------|----------|--------------| +| Asynq Worker Server | `cmd/worker/main.go:140-141`、`240-244` | 消费 Asynq 队列任务 | 适合 | +| PollingConfigManager | `cmd/worker/main.go:143-149` | 加载轮询配置，定时刷新，订阅变更 | 基本可多开 | +| PollingInitializer | `cmd/worker/main.go:156-165` | 启动后从 DB 分批加载全量卡到 Redis 分片队列 | 不适合直接多开 | +| Polling Scheduler | `cmd/worker/main.go:173-188` | 每秒扫描轮询分片队列并投递 Asynq 任务 | 不建议直接多开 | +| Polling activation/data reset | `internal/polling/scheduler.go:225-237` | 每 10 秒做套餐激活检查和流量重置调度 | 不适合直接多开 | +| Asynq Scheduler | `cmd/worker/main.go:200-233` | 周期性注册 `order_expire`、`alert_check`、`data_cleanup`、`daily_traffic_flush` | 不适合直接多开 | + +### 2.2 当前为什么不能直接多开整个 `worker` + +#### 1. Asynq Scheduler 会重复入队 + +`cmd/worker/main.go:200-233` 中每个 `worker` 实例都会启动一个 Asynq Scheduler，并注册： + +- `order_expire` +- `alert_check` +- `data_cleanup` +- `daily_traffic_flush` + +如果启动 3 个完整 `worker` 实例，就会出现： + +- 每分钟不是入队 1 次 `order_expire` +- 而是入队 3 次 `order_expire` + +虽然某些下游任务具备一定幂等性，但会额外增加： + +- 数据库扫描压力 +- Redis 队列压力 +- 日志噪音 +- 故障排查复杂度 + +#### 2. PollingInitializer 会重复全量扫库 + +`cmd/worker/main.go:156-158` 会在启动时执行： + +- `pollingInitializer.StartBackground(ctx)` + +初始化逻辑在 `internal/polling/initializer.go:124-203`，会： + +- 分批读取全量卡数据 +- 写入 Redis 分片轮询队列 +- 写入卡缓存 + +如果多开整个 `worker`，每个实例都会做一遍全量初始化，结果是： + +- 重复扫库 +- 重复写 Redis +- 启动期压力被放大 +- 配置变化时多个实例可能同时重新初始化 + +#### 3. 套餐激活检查会重复扫库和重复入队 + +轮询调度器中的 `processActivationTasks` 位于 `internal/polling/scheduler.go:225-237`，每 10 秒会调用： + +- `HandlePackageActivationCheck(ctx)` +- `HandleDataReset(ctx)` + +其中 `HandlePackageActivationCheck` 在 `internal/polling/package_activation_handler.go:63-109` 中，会主动： + +- 查找过期主套餐 +- 查找孤儿套餐 +- 直接向 Asynq 提交激活任务 + +问题在于这不是“消费已有任务”，而是每个实例自己扫库、自己决定要不要再入队。 + +更重要的是，`enqueueActivationTask` 在 `internal/polling/package_activation_handler.go:469-502` 中并没有使用 Asynq 唯一键控制，因此多实例下会出现： + +- 重复扫描同一批数据 +- 重复提交相同激活任务 +- 下游虽然有 Redis 锁兜底，但前置流量已经放大 + +#### 4. 流量重置调度使用进程内状态，不是分布式单例 + +`internal/polling/data_reset_handler.go:18-20` 中： + +- `lastDailyReset` +- `lastMonthlyReset` +- `lastYearlyReset` + +这三个值是**进程内内存变量**。 + +这意味着： + +- 启动 1 个 `worker`，每分钟只会有 1 份调度检查 +- 启动 3 个 `worker`，每分钟会有 3 份独立调度检查 + +它们会分别调用： + +- `ResetDailyUsage` +- `ResetMonthlyUsage` +- `ResetYearlyUsage` + +相关实现位于 `internal/service/package/reset_service.go:43-240`。当前这里没有单独的分布式 Leader 保护，因此多实例时至少会造成重复扫描，严重时会产生并发处理窗口。 + +#### 5. 轮询分片出队本身虽然相对安全，但不代表整个调度器适合多开 + +轮询分片出队在 `internal/polling/queue_manager.go:24-34` 和 `66-88` 使用 Lua 原子出队： + +- `ZRANGEBYSCORE` +- `ZREM` + +因此就“分片队列抢活”这件事本身，多实例并不是最危险的问题。 + +真正的问题是： + +- 调度器不只做分片出队 +- 还混合了主动扫库、主动定时、主动初始化职责 + +所以风险不是“出队会乱”，而是“整进程多开后单例职责被复制了”。 + +--- + +## 3. 升级目标 + +### 3.1 目标架构 + +升级后的 `worker` 角色拆分如下： + +| 角色 | 职责 | +|------|------| +| `leader` | 运行 Asynq 消费器 + PollingInitializer + PollingScheduler + Asynq Scheduler | +| `consumer` | 只运行 Asynq 消费器，不运行任何调度器和初始化器 | +| `all` | 保持当前兼容模式，单实例部署时继续使用 | + +### 3.2 设计目标 + +1. 保持当前单实例部署兼容 +2. 支持未来多实例扩容 +3. 先做最小改动，不大规模改业务逻辑 +4. 后续可继续升级为“自动主从选主”模式 + +--- + +## 4. 推荐实施方案 + +### 4.1 第一期：先做角色拆分 + +推荐新增配置： + +| 配置项 | 说明 | 默认值 | +|------|------|--------| +| `JUNHONG_WORKER_ROLE` | `all` / `leader` / `consumer` | `all` | +| `JUNHONG_WORKER_INSTANCE_NAME` | 实例名称，便于日志区分 | 空 | + +角色语义定义如下： + +#### `all` + +用于兼容当前行为： + +- 启动 Asynq Worker Server +- 启动 PollingInitializer +- 启动 PollingScheduler +- 启动 Asynq Scheduler + +适用场景： + +- 当前生产单实例 +- 本地开发 +- 首次升级的灰度阶段 + +#### `leader` + +用于未来的单例主 Worker： + +- 启动 Asynq Worker Server +- 启动 PollingInitializer +- 启动 PollingScheduler +- 启动 Asynq Scheduler + +与 `all` 的差别不是运行行为，而是部署语义： + +- `all` 是兼容模式 +- `leader` 是明确表示“该实例承担单例职责” + +#### `consumer` + +用于未来横向扩容的消费实例： + +- 启动 Asynq Worker Server +- **不启动** PollingInitializer +- **不启动** PollingScheduler +- **不启动** Asynq Scheduler + +这样一来： + +- 只有 `leader` 负责“主动调度” +- 所有 `consumer` 只负责“被动消费” + +这就是当前阶段最稳的多实例架构。 + +--- + +## 5. 代码改造点 + +### 5.1 配置层改造 + +需要修改： + +- `pkg/config/config.go` +- `pkg/config/defaults/config.yaml` +- `pkg/config/loader.go` +- `docs/environment-variables.md` + +建议新增： + +```yaml +worker: + role: "all" + instance_name: "" +``` + +并绑定环境变量： + +- `JUNHONG_WORKER_ROLE` +- `JUNHONG_WORKER_INSTANCE_NAME` + +### 5.2 `cmd/worker/main.go` 入口改造 + +将当前入口逻辑改成“共享依赖初始化”和“按角色启停模块”两段。 + +建议改造原则如下： + +#### 永远初始化的部分 + +这些模块无论 `leader` 还是 `consumer` 都需要： + +- Redis +- PostgreSQL +- Storage +- Gateway +- BootstrapWorker +- TaskHandler +- Asynq Worker Server +- PollingConfigManager +- PollingQueueManager +- PollingBase +- LifecycleService + +原因： + +- 消费轮询任务时，`consumer` 仍然需要轮询相关依赖来做重入队、配置读取、停复机处理 + +#### 只在 `all/leader` 启动的部分 + +- `pollingInitializer.StartBackground(ctx)` +- `scheduler.Start(ctx)` +- `asynqScheduler.Run()` + +#### 只在 `all/leader/consumer` 都启动的部分 + +- `workerServer.Run(taskHandler.GetMux())` + +### 5.3 启停日志改造 + +建议在启动时打印： + +- 当前角色 +- 当前实例名 +- 启动了哪些模块 +- 没启动哪些模块 + +例如： + +```text +Worker 角色: consumer +实例名称: worker-consumer-2 +已启用模块: queue_server +已禁用模块: polling_initializer, polling_scheduler, asynq_scheduler +``` + +这样后续排查非常直接。 + +--- + +## 6. 第二期增强方案 + +第一期做完后，已经足够支持“1 主 + N 从”的稳定部署。 + +如果后续希望进一步增强容错，可以继续做以下升级。 + +### 6.1 增加 Leader 分布式锁 + +即使采用 `leader/consumer` 角色，仍然存在人为误配置的风险，例如： + +- 错把两个实例都配置成 `leader` +- 自动扩容时错误复制了 Leader 配置 + +因此第二期建议为以下单例职责加 Redis Leader 锁： + +- Asynq Scheduler 锁 +- Polling Scheduler 锁 +- Polling Initializer 锁 + +推荐思路： + +1. Leader 启动时用 `SET NX EX` 争抢锁 +2. 成功后定时续期 +3. 失去锁时主动停止调度器或直接退出进程 + +这样即使部署层误配，运行时也能收敛到单活。 + +### 6.2 为重复入队场景增加唯一性保护 + +重点场景： + +- `enqueueActivationTask` +- 其他未来可能由调度器主动生成的任务 + +建议做法： + +- 使用 `asynq.Unique(...)` 控制短时间内同一业务键只允许入队一次 +- 或者先用 Redis 锁做业务级去重，再入队 + +这样可以进一步降低重复调度带来的噪音。 + +### 6.3 为流量重置调度加分布式互斥 + +`DataResetHandler` 当前是进程内时钟，不是分布式单例。 + +建议第二期增加： + +- 日重置锁 +- 月重置锁 +- 年重置锁 + +保证无论部署多少个 Leader 候选实例，实际同一时间只有一个实例执行重置扫描。 + +--- + +## 7. 分阶段实施路径 + +### 阶段 1：代码兼容改造 + +目标： + +- 引入 `JUNHONG_WORKER_ROLE` +- 默认保持 `all` +- 单实例行为完全不变 + +验收标准： + +- 不改现有部署配置时，运行行为与当前一致 +- 日志能明确打印当前角色和启用模块 + +### 阶段 2：灰度切换为 `leader` + +目标： + +- 当前生产单实例从 `all` 切成 `leader` +- 验证角色切换不影响现有功能 + +验收标准： + +- 轮询调度器正常启动 +- Asynq Scheduler 正常启动 +- 队列消费正常 + +### 阶段 3：增加第一个 `consumer` + +目标： + +- 新增 1 个 `consumer` +- 验证只有 Leader 运行调度器 +- 验证消费者可共同分担任务处理 + +验收标准： + +- `consumer` 日志中不出现 `轮询调度器已启动` +- `consumer` 日志中不出现 `Asynq Scheduler 已启动` +- 队列堆积下降 +- Redis 心跳键仍只有 Leader 负责刷新 + +### 阶段 4：按需扩容 `consumer` + +目标： + +- 根据任务积压和资源情况逐步增加消费者数量 + +扩容原则： + +- 优先横向扩 `consumer` +- 不随意增加 `leader` +- 先观察队列滞留和 Redis/DB 压力，再决定并发和实例数 + +--- + +## 8. 部署建议 + +### 8.1 当前两台服务器阶段 + +当前建议仍然是： + +- 应用机：`1 个 API + 1 个 worker leader` +- Redis 机：`1 个 Redis` + +此阶段不建议在同一台应用机上同时起多个 `consumer`，因为： + +1. 资源本身有限 +2. 当前最主要瓶颈还不是 Asynq 消费，而是整体调度职责未拆分 + +### 8.2 未来进入多实例阶段 + +推荐部署形态： + +| 节点 | 服务 | +|------|------| +| 应用机 A | `api-1` + `worker-leader` | +| 应用机 B | `api-2` + `worker-consumer-1` | +| 应用机 C | `worker-consumer-2`（可选） | + +如果只有两台应用机，也可以是： + +| 节点 | 服务 | +|------|------| +| 应用机 A | `api-1` + `worker-leader` | +| 应用机 B | `api-2` + `worker-consumer-1` | + +核心原则只有一个： + +- **Leader 固定单例** +- **Consumer 负责横向扩容** + +--- + +## 9. 回滚方案 + +### 9.1 角色切换回滚 + +如果 `leader/consumer` 方案上线后发现问题，回滚顺序建议如下： + +1. 先停止所有 `consumer` +2. 保留一个 `leader` +3. 如果问题仍存在，再将唯一实例切回 `all` + +### 9.2 代码版本回滚 + +如果新版本本身存在问题： + +1. 先停 `consumer` +2. 再停 `leader` +3. 回滚到旧版本单实例 `worker` +4. 只启动 1 个旧版本实例 + +注意： + +- 回滚时不要保留多个旧版本全功能 `worker` 同时运行 +- 否则会重新引入当前这份方案要解决的问题 + +--- + +## 10. 最终建议 + +### 10.1 本次升级推荐做法 + +本次升级建议只做 **第一期最小改动**： + +1. 新增 `worker` 角色概念 +2. 拆出 `leader / consumer / all` +3. 用角色控制模块启停 +4. 暂不一次性引入复杂的自动选主 + +原因： + +1. 改动最小 +2. 风险最低 +3. 能立刻为未来多实例铺路 +4. 当前生产部署也不会被迫大改 + +### 10.2 升级后的长期方向 + +长期建议目标是： + +- `worker-leader` 只负责单例调度职责 +- `worker-consumer` 做水平扩容 +- 第二期再补 Redis Leader 锁和任务唯一性保护 + +这条路径与当前仓库文档里的方向一致，即： + +- Worker 支持多实例 +- 但调度器只保留单例运行 + +相关参考： + +- `specs/002-gorm-postgres-asynq/spec.md:100-103` +- `docs/polling-system/performance-tuning.md:45-66` + +--- + +## 11. 本方案覆盖的直接代码位置 + +为了后续实施时快速定位，当前最关键的修改位置如下： + +| 文件 | 重点修改内容 | +|------|--------------| +| `cmd/worker/main.go` | 按角色拆分模块启停 | +| `pkg/config/config.go` | 新增 Worker 角色配置 | +| `pkg/config/defaults/config.yaml` | 新增默认 Worker 配置 | +| `pkg/config/loader.go` | 绑定 Worker 环境变量 | +| `docs/environment-variables.md` | 补充新环境变量说明 | +| `internal/polling/data_reset_handler.go` | 第二期可补分布式互斥 | +| `internal/polling/package_activation_handler.go` | 第二期可补任务唯一性/去重 | +| `internal/polling/initializer.go` | 第二期可补初始化单例锁 | + +--- + +## 12. 一句话结论 + +当前不是“Worker 永远不能多开”，而是： + +**当前整个 `cmd/worker` 进程里同时混合了可横向扩展的消费职责和必须收敛成单例的调度职责，所以不能直接整进程复制。** + +正确升级路径是： + +**先拆角色，再扩 Consumer，最后补自动单例保护。**