From 3023d47eec53499cae14fe0e1920ac33892cf7d8 Mon Sep 17 00:00:00 2001 From: ryan Date: Thu, 11 Jun 2026 08:50:38 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E8=80=A6=E4=BB=BB=E5=8A=A1=E6=A1=86?= =?UTF-8?q?=E6=9E=B6=E4=B8=8E=E4=B8=9A=E5=8A=A1=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .agent/skills/new-async-task/SKILL.md | 31 +++-- .../references/CODE-EXAMPLES.md | 53 +++----- internal/apps/admin/task/routers.go | 2 +- internal/apps/admin/task/routers_test.go | 14 ++- internal/apps/upload/tasks.go | 20 ++++ internal/apps/user/logics.go | 2 +- internal/apps/user/tasks.go | 47 +++++++- internal/task/constants.go | 113 +----------------- internal/task/handlers/register.go | 11 +- internal/task/meta.go | 92 ++++++++++++++ internal/task/worker/worker.go | 6 +- 11 files changed, 214 insertions(+), 177 deletions(-) create mode 100644 internal/task/meta.go diff --git a/.agent/skills/new-async-task/SKILL.md b/.agent/skills/new-async-task/SKILL.md index 91e0a26b..57b58c3c 100644 --- a/.agent/skills/new-async-task/SKILL.md +++ b/.agent/skills/new-async-task/SKILL.md @@ -12,10 +12,11 @@ description: "项目专用：当新增或修改 Asynq 异步任务、后台任 新增或修改任务前，先快速查看这些文件，确认当前实现没有漂移： - `internal/task/handler.go`: `TaskHandler`、`TaskResult`、可选 `PayloadValidator`。 -- `internal/task/constants.go`: Asynq 任务类型常量、Admin 任务类型常量、`TaskMeta`、`TaskParam`、`DispatchableTasks`。 +- `internal/task/constants.go`: 框架通用常量，如 `QueueDefault` 和 `DefaultMaxRetry`。 +- `internal/task/meta.go`: 框架任务元数据结构体（TaskParam、TaskMeta）及全局动态注册与查询接口。 - `internal/task/executor.go`: `RegisterHandler`、`ValidateAndNormalizePayload`、`DispatchTask`、`RetryTask`、`ProcessTask`、`AppendLog`。 -- `internal/task/handlers/register.go`: 内置 handler 的统一注册点，Admin API 和 Worker 都依赖它。 -- `internal/task/worker/worker.go`: Asynq mux 路由和队列配置。 +- `internal/task/handlers/register.go`: 内置 handler 和元数据的统一注册点，Admin API 和 Worker 都依赖它。 +- `internal/task/worker/worker.go`: Asynq mux 动态路由分发和队列配置。 - `internal/task/scheduler/scheduler.go`: Cron 调度。 - `internal/apps/admin/task/routers.go`: Admin 下发、查询、详情、重试 API。 - 现有参考：`internal/apps/upload/tasks.go`（无参数任务）、`internal/apps/user/tasks.go`（带参数任务）。 @@ -34,10 +35,9 @@ Admin dispatch -> ValidateAndNormalizePayload -> DispatchTask > 需要可复制的代码模板时，阅读 [references/CODE-EXAMPLES.md](references/CODE-EXAMPLES.md)。那里包含任务常量、无参数 handler、带参数 `PayloadValidator`、统一注册、Worker 路由、Cron 配置和测试示例。 -1. 定义任务元数据。 - - 在 `internal/task/constants.go` 添加 Asynq task type 常量，例如 `upload:cleanup_unused`。 - - 添加 Admin 可下发 task type 常量，例如 `cleanup_unused_uploads`。 - - 在 `DispatchableTasks` 添加 `TaskMeta`，设置 `AsynqTask`、`Name`、`Description`、`MaxRetry`、`Queue`、`Retryable`。 +1. 定义任务元数据与常量。 + - 业务包常量与元数据：在对应业务模块的 `internal/apps//tasks.go` 中定义 Asynq 任务类型常量（如 `CleanupUnusedUploadsTask = "upload:cleanup_unused"`）和 Admin 任务类型常量（如 `TaskTypeCleanupUploads = "cleanup_unused_uploads"`）。 + - 在同一 `tasks.go` 文件中定义该任务的 `TaskMeta` 元数据变量（如 `CleanupUnusedUploadsMeta = task.TaskMeta{...}`），配置 `Type`、`AsynqTask`、`Name`、`Description`、`MaxRetry`、`Queue`、`Retryable` 等字段。 - 有参数任务在 `Params` 中描述前端表单字段。`TaskParam.Name` 必须与 payload JSON tag 对齐。 2. 实现 handler。 @@ -48,19 +48,16 @@ Admin dispatch -> ValidateAndNormalizePayload -> DispatchTask - 不要在 handler 中写复杂 SQL；复杂查询放到 `internal/model/` 或 `internal/service/`。 - 新增 Go 文件后检查 license header；必要时运行 `make license`。 -3. 统一注册 handler。 - - 在 `internal/task/handlers/register.go` 导入业务模块并调用 `task.RegisterHandler(asynqTaskType, handler)`。 - - 这里是 Admin payload 校验和 Worker 执行共同依赖的注册点。不要只在 worker 包里注册。 +3. 统一注册 handler 与元数据。 + - 在 `internal/task/handlers/register.go` 导入业务模块，调用 `task.RegisterHandler(asynqTaskType, handler)` 注册处理器。 + - 同时，在该文件中调用 `task.RegisterTaskMeta(meta)` 注册刚才在业务模块中定义的任务元数据。 + - 这里是 Admin 校验、元数据获取和 Worker 执行共同依赖的注册点。 -4. 注册 Worker 路由。 - - 在 `internal/task/worker/worker.go` 的 Asynq mux 中添加 `mux.HandleFunc(task.YourAsynqTask, task.ProcessTask)`。 - - 所有业务任务都应交给 `task.ProcessTask`，由 executor 根据 task type 分发到 handler。 - -5. 如需 Cron 调度，系统默认定时任务必须通过 SQL 迁移（goose）初始化。 - - 确保任务在 `DispatchableTasks` 中已正确配置 `TaskMeta`。 +4. 如需 Cron 调度，系统默认定时任务必须通过 SQL 迁移（goose）初始化。 + - 确保任务已正确注册并载入全局元数据池中。 - 在 `internal/db/migrator/goose/postgres` 和 `sqlite` 下编写 migration 脚本，使用 `INSERT INTO schedules` 语句初始化任务，指定 `task_type` 和 `cron` 等字段。必须妥善处理冲突（如 `ON CONFLICT DO NOTHING`）以支持幂等。 -6. 如改动 Admin API。 +5. 如改动 Admin API。 - handler 放在 `internal/apps/admin//` 或现有 Admin task 模块内。 - 路由只在 `internal/router/router.go` 注册。 - 响应保持 `{ "error_msg": "", "data": ... }`，分页保持 `{ "total": 0, "results": [] }`。 diff --git a/.agent/skills/new-async-task/references/CODE-EXAMPLES.md b/.agent/skills/new-async-task/references/CODE-EXAMPLES.md index f4758b07..ffa77c0f 100644 --- a/.agent/skills/new-async-task/references/CODE-EXAMPLES.md +++ b/.agent/skills/new-async-task/references/CODE-EXAMPLES.md @@ -2,28 +2,33 @@ 这些示例用于新增或修改 Wavelet Asynq 任务时快速套用。复制前先对照当前代码，因为任务框架可能随项目演进。 -## 任务元数据 +## 任务元数据与常量定义 -在 `internal/task/constants.go` 添加 Asynq task type、Admin task type 和 `TaskMeta`。 +在对应的业务包 `internal/apps//tasks.go` 中定义 Asynq task type、Admin task type 和 `TaskMeta`。 ```go +package upload + +import ( + "github.com/Rain-kl/Wavelet/internal/task" +) + // 异步任务类型标识。格式建议为 "{module}:{action}"。 const CleanupUnusedUploadsTask = "upload:cleanup_unused" // 管理员可下发的任务类型标识。用于 Admin API 的 task_type。 const TaskTypeCleanupUploads = "cleanup_unused_uploads" -var DispatchableTasks = []TaskMeta{ - { - Type: TaskTypeCleanupUploads, - AsynqTask: CleanupUnusedUploadsTask, - Name: "清理未使用上传", - Description: "清理超过1小时未使用的上传文件", - SupportsTime: false, - MaxRetry: defaultMaxRetry, - Queue: QueueDefault, - Retryable: true, - }, +// CleanupUnusedUploadsMeta 任务元数据 +var CleanupUnusedUploadsMeta = task.TaskMeta{ + Type: TaskTypeCleanupUploads, + AsynqTask: CleanupUnusedUploadsTask, + Name: "清理未使用上传", + Description: "清理超过1小时未使用的上传文件", + SupportsTime: false, + MaxRetry: task.DefaultMaxRetry, + Queue: task.QueueDefault, + Retryable: true, } ``` @@ -175,28 +180,6 @@ func Register() { } ``` -## Worker 路由 - -在 `internal/task/worker/worker.go` 的 mux 上添加任务类型。所有业务任务都指向 `task.ProcessTask`。 - -```go -func StartWorker() error { - asynqServer := asynq.NewServer(task.RedisOpt, asynq.Config{ - Concurrency: config.Config.Worker.Concurrency, - ShutdownTimeout: workerShutdownTimeout, - Queues: buildQueuesFromConfig(), - StrictPriority: config.Config.Worker.StrictPriority, - }) - - mux := asynq.NewServeMux() - mux.Use(taskLoggingMiddleware) - mux.HandleFunc(task.CleanupUnusedUploadsTask, task.ProcessTask) - mux.HandleFunc(task.SendEmailTask, task.ProcessTask) - - return asynqServer.Run(mux) -} -``` - ## Cron 调度和配置 系统默认的定时任务必须通过 Goose SQL 迁移初始化插入到 `schedules` 表。 diff --git a/internal/apps/admin/task/routers.go b/internal/apps/admin/task/routers.go index 54c0b9a8..d9520a00 100644 --- a/internal/apps/admin/task/routers.go +++ b/internal/apps/admin/task/routers.go @@ -37,7 +37,7 @@ func init() { // @Failure 403 {object} util.ResponseAny "无管理员权限" // @Router /api/v1/admin/tasks/types [get] func ListTaskTypes(c *gin.Context) { - c.JSON(http.StatusOK, util.OK(task.DispatchableTasks)) + c.JSON(http.StatusOK, util.OK(task.GetDispatchableTasks())) } // DispatchTaskRequest 下发任务请求 diff --git a/internal/apps/admin/task/routers_test.go b/internal/apps/admin/task/routers_test.go index b0a54f87..5a0a60bc 100644 --- a/internal/apps/admin/task/routers_test.go +++ b/internal/apps/admin/task/routers_test.go @@ -15,6 +15,8 @@ import ( "time" "github.com/Rain-kl/Wavelet/internal/apps/oauth" + "github.com/Rain-kl/Wavelet/internal/apps/upload" + "github.com/Rain-kl/Wavelet/internal/apps/user" "github.com/Rain-kl/Wavelet/internal/model" "github.com/Rain-kl/Wavelet/internal/task" "github.com/Rain-kl/Wavelet/internal/testhelper" @@ -88,13 +90,13 @@ func TestListTaskTypes(t *testing.T) { foundCleanup := false for _, m := range taskMetas { - if m.Type == task.TaskTypeCleanupUploads { + if m.Type == upload.TaskTypeCleanupUploads { foundCleanup = true break } } if !foundCleanup { - t.Errorf("expected task type %s to be listed", task.TaskTypeCleanupUploads) + t.Errorf("expected task type %s to be listed", upload.TaskTypeCleanupUploads) } } @@ -107,7 +109,7 @@ func TestDispatchTask(t *testing.T) { t.Run("dispatch valid task successfully", func(t *testing.T) { payload := DispatchTaskRequest{ - TaskType: task.TaskTypeCleanupUploads, + TaskType: upload.TaskTypeCleanupUploads, } body, _ := json.Marshal(payload) req, _ := http.NewRequest("POST", "/api/v1/admin/tasks/dispatch", bytes.NewBuffer(body)) @@ -130,7 +132,7 @@ func TestDispatchTask(t *testing.T) { t.Run("dispatch send_email task successfully with valid payload", func(t *testing.T) { payload := DispatchTaskRequest{ - TaskType: task.TaskTypeSendEmail, + TaskType: user.TaskTypeSendEmail, Payload: `{"to":"receiver@example.com","subject":"Test Subject","body":"Test Body"}`, } body, _ := json.Marshal(payload) @@ -149,7 +151,7 @@ func TestDispatchTask(t *testing.T) { t.Run("dispatch send_email task failure with invalid payload json", func(t *testing.T) { payload := DispatchTaskRequest{ - TaskType: task.TaskTypeSendEmail, + TaskType: user.TaskTypeSendEmail, Payload: `{"to":`, } body, _ := json.Marshal(payload) @@ -166,7 +168,7 @@ func TestDispatchTask(t *testing.T) { t.Run("dispatch send_email task failure with missing fields", func(t *testing.T) { payload := DispatchTaskRequest{ - TaskType: task.TaskTypeSendEmail, + TaskType: user.TaskTypeSendEmail, Payload: `{"to":"","subject":"Test","body":"Test"}`, } body, _ := json.Marshal(payload) diff --git a/internal/apps/upload/tasks.go b/internal/apps/upload/tasks.go index a9079074..7cf7da4b 100644 --- a/internal/apps/upload/tasks.go +++ b/internal/apps/upload/tasks.go @@ -16,6 +16,26 @@ import ( "gorm.io/gorm" ) +// 异步任务名称与管理类型定义 +const ( + // CleanupUnusedUploadsTask 清理未使用上传任务标识 + CleanupUnusedUploadsTask = "upload:cleanup_unused" + // TaskTypeCleanupUploads 清理未使用上传管理类型 + TaskTypeCleanupUploads = "cleanup_unused_uploads" +) + +// CleanupUnusedUploadsMeta represents the task metadata. +var CleanupUnusedUploadsMeta = task.TaskMeta{ + Type: TaskTypeCleanupUploads, + AsynqTask: CleanupUnusedUploadsTask, + Name: "清理未使用上传", + Description: "清理超过1小时未使用的上传文件", + SupportsTime: false, + MaxRetry: task.DefaultMaxRetry, + Queue: task.QueueDefault, + Retryable: true, +} + // CleanupUnusedUploadsHandler 清理未使用上传文件的异步任务处理器 type CleanupUnusedUploadsHandler struct{} diff --git a/internal/apps/user/logics.go b/internal/apps/user/logics.go index a6450b54..851161ed 100644 --- a/internal/apps/user/logics.go +++ b/internal/apps/user/logics.go @@ -109,7 +109,7 @@ func sendEmailVerificationCode(ctx context.Context, email, scene, templateName s Body: emailBody, } payloadBytes, _ := json.Marshal(payload) - _, err = task.DispatchTask(ctx, task.TaskTypeSendEmail, payloadBytes, "system") + _, err = task.DispatchTask(ctx, TaskTypeSendEmail, payloadBytes, "system") if err != nil { return errors.New(errDispatchEmailTaskFailed) } diff --git a/internal/apps/user/tasks.go b/internal/apps/user/tasks.go index 178e482f..0c26a216 100644 --- a/internal/apps/user/tasks.go +++ b/internal/apps/user/tasks.go @@ -1,4 +1,3 @@ -// Copyright 2025 linux.do // Copyright 2026 Arctel.net // SPDX-License-Identifier: Apache-2.0 @@ -17,6 +16,52 @@ import ( "github.com/Rain-kl/Wavelet/internal/util/mail" ) +// 异步任务名称与管理类型定义 +const ( + // SendEmailTask 发送邮件任务标识 + SendEmailTask = "mail:send" + // TaskTypeSendEmail 发送邮件管理类型 + TaskTypeSendEmail = "send_email" +) + +// SendEmailMeta represents the task metadata. +var SendEmailMeta = task.TaskMeta{ + Type: TaskTypeSendEmail, + AsynqTask: SendEmailTask, + Name: "发送邮件", + Description: "异步发送系统邮件", + SupportsTime: false, + MaxRetry: task.DefaultMaxRetry, + Queue: task.QueueDefault, + Retryable: true, + Params: []task.TaskParam{ + { + Name: "to", + Label: "接收邮箱 (To)", + Type: "string", + Required: true, + Placeholder: "receiver@example.com", + Description: "接收邮件的目标邮箱地址", + }, + { + Name: "subject", + Label: "邮件主题 (Subject)", + Type: "string", + Required: true, + Placeholder: "请输入邮件主题", + Description: "发送邮件的主题标题", + }, + { + Name: "body", + Label: "邮件内容 (Body)", + Type: "text", + Required: true, + Placeholder: "请输入邮件内容（支持 HTML 格式）", + Description: "发送邮件的内容主体", + }, + }, +} + // SendEmailPayload 邮件发送任务载荷 type SendEmailPayload struct { To string `json:"to"` diff --git a/internal/task/constants.go b/internal/task/constants.go index 7bd9ccfa..4ad07259 100644 --- a/internal/task/constants.go +++ b/internal/task/constants.go @@ -5,119 +5,10 @@ // Package task 定义异步任务类型与调度常量 package task -// 异步任务类型标识 -const ( - CleanupUnusedUploadsTask = "upload:cleanup_unused" - SendEmailTask = "mail:send" -) - // 任务队列名称 const ( QueueDefault = "default" ) -// 管理员可下发的任务类型标识 -const ( - TaskTypeCleanupUploads = "cleanup_unused_uploads" - TaskTypeSendEmail = "send_email" -) - -// defaultMaxRetry 任务默认最大重试次数 -const defaultMaxRetry = 3 - -// TaskParam 任务参数定义 -// -//nolint:revive // TaskParam 保留完整名称以避免与通用 Param 混淆 -type TaskParam struct { - Name string `json:"Name"` // 参数键名 - Label string `json:"Label"` // 显示名称 - Type string `json:"Type"` // 类型：string, text, number, boolean - Required bool `json:"Required"` // 是否必填 - Placeholder string `json:"Placeholder"` // 占位符 - Description string `json:"Description"` // 描述 -} - -// TaskMeta 任务元数据 -// -//nolint:revive // TaskMeta 保留完整名称以避免与通用 Meta 混淆 -type TaskMeta struct { - Type string - AsynqTask string - Name string - Description string - SupportsTime bool - MaxRetry int - Queue string - Retryable bool // 是否支持手动重试 - Params []TaskParam -} - -// DispatchableTasks 可下发的任务列表 -var DispatchableTasks = []TaskMeta{ - { - Type: TaskTypeCleanupUploads, - AsynqTask: CleanupUnusedUploadsTask, - Name: "清理未使用上传", - Description: "清理超过1小时未使用的上传文件", - SupportsTime: false, - MaxRetry: defaultMaxRetry, - Queue: QueueDefault, - Retryable: true, - }, - { - Type: TaskTypeSendEmail, - AsynqTask: SendEmailTask, - Name: "发送邮件", - Description: "异步发送系统邮件", - SupportsTime: false, - MaxRetry: defaultMaxRetry, - Queue: QueueDefault, - Retryable: true, - Params: []TaskParam{ - { - Name: "to", - Label: "接收邮箱 (To)", - Type: "string", - Required: true, - Placeholder: "receiver@example.com", - Description: "接收邮件的目标邮箱地址", - }, - { - Name: "subject", - Label: "邮件主题 (Subject)", - Type: "string", - Required: true, - Placeholder: "请输入邮件主题", - Description: "发送邮件的主题标题", - }, - { - Name: "body", - Label: "邮件内容 (Body)", - Type: "text", - Required: true, - Placeholder: "请输入邮件内容（支持 HTML 格式）", - Description: "发送邮件的内容主体", - }, - }, - }, -} - -// GetTaskMeta 根据任务类型获取元数据 -func GetTaskMeta(taskType string) *TaskMeta { - for _, t := range DispatchableTasks { - if t.Type == taskType { - return &t - } - } - return nil -} - -// GetTaskMetaByAsynqTask 根据 Asynq 任务名称获取元数据 -func GetTaskMetaByAsynqTask(asynqTask string) *TaskMeta { - for _, t := range DispatchableTasks { - if t.AsynqTask == asynqTask { - return &t - } - } - return nil -} +// DefaultMaxRetry 任务默认最大重试次数 +const DefaultMaxRetry = 3 diff --git a/internal/task/handlers/register.go b/internal/task/handlers/register.go index 94101cde..f980e485 100644 --- a/internal/task/handlers/register.go +++ b/internal/task/handlers/register.go @@ -11,8 +11,13 @@ import ( "github.com/Rain-kl/Wavelet/internal/task" ) -// Register registers all built-in task handlers. +// Register registers all built-in task handlers and their metadata. func Register() { - task.RegisterHandler(task.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{}) - task.RegisterHandler(task.SendEmailTask, &user.SendEmailHandler{}) + // upload + task.RegisterHandler(upload.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{}) + task.RegisterTaskMeta(upload.CleanupUnusedUploadsMeta) + + // user + task.RegisterHandler(user.SendEmailTask, &user.SendEmailHandler{}) + task.RegisterTaskMeta(user.SendEmailMeta) } diff --git a/internal/task/meta.go b/internal/task/meta.go new file mode 100644 index 00000000..fb16b005 --- /dev/null +++ b/internal/task/meta.go @@ -0,0 +1,92 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package task + +import ( + "sync" +) + +// TaskParam 任务参数定义 +// +//nolint:revive // TaskParam 保留完整名称以避免与通用 Param 混淆 +type TaskParam struct { + Name string `json:"Name"` // 参数键名 + Label string `json:"Label"` // 显示名称 + Type string `json:"Type"` // 类型：string, text, number, boolean + Required bool `json:"Required"` // 是否必填 + Placeholder string `json:"Placeholder"` // 占位符 + Description string `json:"Description"` // 描述 +} + +// TaskMeta 任务元数据 +// +//nolint:revive // TaskMeta 保留完整名称以避免与通用 Meta 混淆 +type TaskMeta struct { + Type string + AsynqTask string + Name string + Description string + SupportsTime bool + MaxRetry int + Queue string + Retryable bool // 是否支持手动重试 + Params []TaskParam +} + +var ( + dispatchableTasksMutex sync.RWMutex + dispatchableTasks []TaskMeta +) + +// RegisterTaskMeta 注册任务元数据到全局列表 +func RegisterTaskMeta(meta TaskMeta) { + dispatchableTasksMutex.Lock() + defer dispatchableTasksMutex.Unlock() + dispatchableTasks = append(dispatchableTasks, meta) +} + +// GetDispatchableTasks 获取所有已注册的元数据列表（返回副本以避免并发并发读写冲突） +func GetDispatchableTasks() []TaskMeta { + dispatchableTasksMutex.RLock() + defer dispatchableTasksMutex.RUnlock() + + metas := make([]TaskMeta, len(dispatchableTasks)) + copy(metas, dispatchableTasks) + return metas +} + +// GetTaskMeta 根据任务类型获取元数据 +func GetTaskMeta(taskType string) *TaskMeta { + dispatchableTasksMutex.RLock() + defer dispatchableTasksMutex.RUnlock() + for _, t := range dispatchableTasks { + if t.Type == taskType { + copied := t + return &copied + } + } + return nil +} + +// GetTaskMetaByAsynqTask 根据 Asynq 任务名称获取元数据 +func GetTaskMetaByAsynqTask(asynqTask string) *TaskMeta { + dispatchableTasksMutex.RLock() + defer dispatchableTasksMutex.RUnlock() + for _, t := range dispatchableTasks { + if t.AsynqTask == asynqTask { + copied := t + return &copied + } + } + return nil +} + +// GetRegisteredAsynqTasks 返回所有已注册的 Asynq 任务名称，以便动态注册路由 +func GetRegisteredAsynqTasks() []string { + keys := make([]string, 0, len(handlerRegistry)) + for k := range handlerRegistry { + keys = append(keys, k) + } + return keys +} diff --git a/internal/task/worker/worker.go b/internal/task/worker/worker.go index 7026afd3..72e2668d 100644 --- a/internal/task/worker/worker.go +++ b/internal/task/worker/worker.go @@ -39,8 +39,10 @@ func StartWorker() error { // 统一使用 task.ProcessTask 处理所有任务类型 // 框架内部自动分发到对应的 TaskHandler 实现 - mux.HandleFunc(task.CleanupUnusedUploadsTask, task.ProcessTask) - mux.HandleFunc(task.SendEmailTask, task.ProcessTask) + // 动态注册所有已注册的任务处理器路由，框架内部自动分发到对应的 TaskHandler 实现 + for _, taskName := range task.GetRegisteredAsynqTasks() { + mux.HandleFunc(taskName, task.ProcessTask) + } // 启动服务器 return asynqServer.Run(mux)