fwhQueueNew 设计文档

概览

fwhQueueNew 是使用 Asynq 框架重写的 FWH 任务队列，替代原有基于 Redis Stream 手动实现的 fwhQueue。

重写目标

消除手动状态管理带来的 bug（如 .String() vs .Hex() 导致任务状态不一致）
统一任务监控（所有任务可在 Asynqmon Web UI 中查看）
简化重试逻辑（利用 Asynq 内置重试机制替代手动 for 循环）
提高可维护性（声明式任务配置，减少状态同步代码）

目录结构

task/queue/fwhQueueNew/
├── fwhTaskQueue.go              # 初始化：Server、Client、Inspector、常量定义
├── fwhTaskBatchDownload.go      # BatchDownload 主任务 handler
├── fwhTaskCallFile.go           # CallFile 子任务 handler（EmsLog + FaultRecord）
├── fwhTaskSiteAnalysis.go       # SiteAnalysis 任务 handler
├── fwhTaskShutdownStatistics.go # ShutdownStatistics 任务 handler
├── fwhTaskSiteInfoUpdate.go     # SiteInfoUpdate 任务 handler
├── fwhEnqueue.go                # 对外暴露的入队函数
└── DESIGN.md                    # 本文档

架构设计

队列拓扑

┌─────────────────────────────────────────────────────────┐
│                      Asynq Server                        │
│                     Concurrency: 6                        │
│                                                          │
│  ┌──────────────────┐    ┌───────────────────────────┐  │
│  │  fwh:main (2)    │    │  fwh:download (4)         │  │
│  │                  │    │                           │  │
│  │ • BatchDownload  │    │ • CallFile (EmsLog)       │  │
│  │ • SiteAnalysis   │    │ • CallFile (FaultRecord)  │  │
│  │ • ShutdownStats  │    │                           │  │
│  │ • SiteInfoUpdate │    │                           │  │
│  └──────────────────┘    └───────────────────────────┘  │
└─────────────────────────────────────────────────────────┘

fwh:main：主队列，2 个 worker，处理业务编排类任务
fwh:download：下载队列，4 个 worker，处理文件下载类任务（可并发）

任务类型

任务类型	常量	队列	重试	超时
批量数据下载	`fwh:BatchDownload`	main	0	120min
文件下载子任务	`fwh:CallFile`	download	3	35min
站点分析	`fwh:SiteAnalysis`	main	0	60min
Shutdown统计	`fwh:ShutdownStatistics`	main	0	30min
站点信息更新	`fwh:SiteInfoUpdate`	main	0	60min

BatchDownload 子任务协调机制

原有方案（问题）

旧方案：Redis Set (standby/failed) + Redis Stream per SN
问题：
1. mainTaskId 使用 .String() vs .Hex() 不一致导致 key 不匹配
2. 提前 return 的路径忘记更新 standby → 主任务永远等待
3. 手动 goroutine per SN 的生命周期管理复杂

新方案

新方案：Redis 计数器 + 失败原因列表

入队时：
  SET fwh:callfile:counter:{mainTaskId} = N (子任务总数)

子任务完成时（无论成功/失败）：
  DECR fwh:callfile:counter:{mainTaskId}
  如果失败：LPUSH fwh:callfile:failures:{mainTaskId} "失败原因"

主任务轮询：
  GET fwh:callfile:counter:{mainTaskId}
  当 counter <= 0 时：
    LRANGE fwh:callfile:failures:{mainTaskId} 0 -1 → 收集失败原因
    DEL counter, failures → 清理

优势：

原子操作（DECR/LPUSH），无竞态
不依赖子任务的执行路径，无论 handler 怎么 return/panic，Asynq 都会管理任务生命周期
结构简单，只有 2 个 Redis key

重试策略

CallFile 子任务

Asynq 外层重试（MaxRetry=3）：
  - 第1次失败 → 等待 30s 重试
  - 第2次失败 → 等待 2min 重试
  - 第3次失败 → 标记为最终失败，报告失败原因

内部下载重试（maxCallRetry=3）：
  - 针对 HTTP 下载的短暂网络错误
  - 不涉及重新召唤，只重试下载动作

其他任务

业务错误（数据格式错误、API 返回失败等）不重试，handler 返回 nil 避免 Asynq 重试。只有基础设施错误（Redis 断连等）才返回 error 触发重试。

与旧 fwhQueue 的对比

维度	fwhQueue (旧)	fwhQueueNew (新)
底层	手动 Redis Stream	Asynq（内部用 Redis List + SortedSet）
并发控制	单 worker 顺序 + 手动 goroutine per SN	`Concurrency` + `Queues` 声明式
重试	`XAutoClaim` + `XPendingExt` 手动计数	`MaxRetry(N)` + 指数退避
任务去重	手动 Redis 分布式锁	`asynq.Unique(duration)`
任务状态	内存变量`runningID`/`failedTaskList`	Asynq 自动管理
子任务协调	Redis Set (standby/failed) 手动移动	Redis 计数器 DECR
监控	手动`GetTaskInfoList`	Asynqmon Web UI
Bug 风险	高（多种状态同步路径）	低（结构化保证）
代码量	~800 行（fwhTaskQueue.go + callFwhFile.go）	~500 行（分布在 7 个文件）

迁移指南

需要修改的文件

1. `task/queue/taskQueue.go`

package taskQueue

import (
    task_core "fwh-toolbox-server/task/core"
    "fwh-toolbox-server/task/queue/fwhQueueNew"  // 替换 fwhQueue
    "fwh-toolbox-server/task/queue/simCardQueue"
    "github.com/hibiken/asynq"
)

func InitTaskQueue(tc *task_core.TaskCore, redisOpt asynq.RedisClientOpt) {
    if tc == nil {
        panic("InitTaskQueue failed, TaskCore is nil.")
    }

    fwhQueueNew.InitFwhQueueNew(tc, redisOpt)  // 替换 fwhQueue.InitFwhTaskQueue()
    simCardQueue.InitSimCardTaskQueue(tc)
}

2. `task/task.go`

package task

import (
    task_core "fwh-toolbox-server/task/core"
    taskQueue "fwh-toolbox-server/task/queue"
    "github.com/hibiken/asynq"
)

func InitTask(redisOpt TaskRedisOpt) {
    redisClientOpt := asynq.RedisClientOpt{
        Addr:     redisOpt.Addr,
        DB:       redisOpt.DB,
        Username: redisOpt.Username,
        Password: redisOpt.Password,
    }

    tc := task_core.NewTaskCore(redisClientOpt)
    taskQueue.InitTaskQueue(tc, redisClientOpt)  // 传入 redisOpt
}

3. `services/batchDownloadServices.go`

// 将：
import "fwh-toolbox-server/task/queue/fwhQueue"
fwhQueue.EnqueueBatchDownloadTask(...)

// 替换为：
import "fwh-toolbox-server/task/queue/fwhQueueNew"
fwhQueueNew.EnqueueBatchDownloadTask(...)

4. 其他引用 `fwhQueue.EnqueueXxx` 的地方

全局搜索 fwhQueue.Enqueue 替换为 fwhQueueNew.Enqueue：

services/sitesAnalysisServices.go
scheduler/scheduler.go（如果有定时触发 ShutdownStatistics）
controllers/ 中直接调用入队的地方

5. `main.go` 中的 Asynqmon 配置

无需改动，Asynqmon 已经指向同一个 Redis DB，新队列的任务会自动出现在 UI 中。

注意事项

Redis DB：新队列使用与 Asynq Server 相同的 Redis DB（当前为 DB 12），计数器 key 也在同一个 DB 中
并行安全：fwh:download 队列的 4 个 worker 可能同时处理不同 SN 的文件下载，这是安全的（每个 SN 独立）
超时保护：所有任务都设置了 Timeout，防止 handler 卡死
历史保留：Retention(7*24*time.Hour) 保留已完成任务 7 天，方便在 Asynqmon 中查看历史
平滑迁移：新旧队列可以共存一段时间，逐步将调用切换到新队列

DESIGN.md 8.5 KB Historia Czysty