stock-analysis/.trae/documents/async-task-system.md

异步任务系统 + 并发拉取批量写入

Context

当前数据入库通过 CLI cmd/ingest 串行执行，全量 1035 只转债约需 1 小时。用户要求：

1. Web 控制页增加全量/增量更新按钮，显示数据库数据时间
2. 后端引入 Redis，异步执行入库任务，任务状态实时写 Redis
3. 前端轮询后端接口实时刷新任务状态
4. 任务和结果同时持久化到 SQLite，Web 可查询任务列表
5. 核心优化：并发拉取（多 worker 同时从 API 取数据）+ 批量写入（20 只一批统一写 SQLite），替代串行模式

目标：全量入库从 1 小时降至约 5 分钟，且支持 Web 端操作。

架构概览

前端(轮询2s) → POST /api/tasks → TaskController → JobManager.Start()
                                                      ↓ (异步 goroutine)
                                            Runner: N workers 并发拉取
                                                      ↓ resultCh
                                            Collector: 单 goroutine 批量写 SQLite
                                                      ↓ 每20只
                                            Store.SaveKlinesBatch (单事务)
                                                      ↓
                                            Redis(实时进度) + SQLite(持久化)

关键约束与对策：

• sqlite.Store.SetMaxOpenConns(1) → 单一 collector goroutine 串行写入（worker 不直接写 DB）
• eastmoney.Client.throttle() 非线程安全（lastReq 无锁）→ 加 sync.Mutex + 可配置间隔，job 用 NewClientWithThrottle(0) 关闭内部节流，由全局限速器统一控制
• 预加载优化：ListCodes() + LastKlineDates()（新增 GROUP BY 一次查询）替代 600 次逐只查询，避免占用唯一连接堵塞写入
• MarketController 每请求 sqlite.Open → 改为注入共享 *sqlite.Store，消除连接池争用

实现步骤

阶段 A：基础设施

1. 修改 internal/datasource/eastmoney/client.go：Client 加 sync.Mutex + interval time.Duration 字段；throttle() 加锁；新增 NewClientWithThrottle(interval) 构造函数（NewClient() 行为不变，默认 2s）

2. 新增依赖：go get github.com/redis/go-redis/v9，提升 github.com/google/uuid 为直接依赖

3. 新建 internal/cache/cache.go：Task 结构体（ID/Type/Status/Total/Done/Success/Failed/FailedCodes/Error/时间戳）、Cache 接口（GetTask/SetTask/DeleteTask/ListActiveTaskIDs/AddActiveTask/RemoveActiveTask/Ping）、NoopCache 空实现（Redis 降级用）

4. 新建 internal/cache/redis.go：RedisCache 实现，键设计 stock-analysis:task:{id} → Task JSON（TTL 24h）、stock-analysis:tasks:active → SET

阶段 B：存储层扩展

5. 修改 internal/storage/sqlite/store.go：

   • schemaSQL 追加 ingest_task 表（id/type/status/total/done/success/failed/failed_codes/params/error/started_at/finished_at）
   • 新增 SaveKlinesBatch(ctx, batch map[string][]model.Kline)：单事务批量写入多只转债K线，带 3 次重试
   • 新增 LastKlineDates(ctx) map[string]string：SELECT code, MAX(date) FROM convertible_kline GROUP BY code 一次查询

6. 新建 internal/storage/sqlite/task.go：任务 CRUD（SaveTask/UpdateTaskProgress/FinishTask/GetTask/ListTasks），ListTasks 按 started_at DESC 排序

阶段 C：任务执行核心

7. 新建 internal/job/store.go：定义 Store 接口（含 K线批量写 + 任务 CRUD），编译期断言 *sqlite.Store 实现它

8. 新建 internal/job/ingest.go：resolveCandidates（full 走 eastmoney datacenter、incremental 走 store.ListCodes）、yearOf、resolveIncrementalBeg（从 cmd/ingest 提取）

9. 新建 internal/job/runner.go：核心并发逻辑

   • run()：预加载 skipSet/lastDateMap → 启动 collector goroutine → N worker 并发拉取 → 等待完成 → flush 残余 batch
   • collect()：单 goroutine，从 resultCh 收集，每 20 只调 SaveKlinesBatch，调 reportProgress
   • fetchOne()：单只转债拉取（全量用 FetchFullKline，增量用 FetchIncrementalKline）
   • reportProgress()：每次写 Redis，每 5 只写 SQLite（减少写压力）

10. 新建 internal/job/job.go：Manager 管理任务注册表 + 全局限速器 gate（令牌桶：400ms 间隔，突发 4）

    • Start(ctx, type, since, limit)：创建 Task → 存 SQLite+Redis → 异步启动 Runner
    • Cancel(id)：调用 context.CancelFunc
    • Get(ctx, id)：Redis 优先，回退 SQLite
    • List(ctx, limit)：SQLite 历史 + Redis 实时覆盖运行中任务
    • StopAll()：服务关闭时优雅停止

阶段 D：HTTP 层

11. 新建 internal/controller/task.go：TaskController，4 个 handler

    • POST /api/tasks：创建任务 {"type":"full"|"incremental","since":"","limit":0}
    • GET /api/tasks：任务列表
    • GET /api/tasks/{id}：任务状态
    • POST /api/tasks/{id}/cancel：取消任务

12. 修改 internal/controller/api.go：MarketController 字段 dbPath string → store *sqlite.Store；NewMarketController(svc, store)；移除 handler 中临时 sqlite.Open

13. 修改 internal/router/router.go：New 签名增加 tc *TaskController；注册 4 条 task 路由（均 requireAuth）

14. 修改 cmd/server/main.go：

    • 共享 sqlite.Open(dbPath)（替代 controller 临时 open）
    • initCache()：读 REDIS_ADDR（默认 localhost:6379），连不上降级 NoopCache
    • initJobManager()：创建无节流 eastmoney client + fallback + Manager
    • defer mgr.StopAll()
    • Docker 启动 Redis：docker run -d --name stock-redis -p 6379:6379 redis:7-alpine

阶段 E：前端

15. 修改 web/style.css：新增 .progress-bar/.progress-fill/.badge-*（running/success/failed/cancelled）样式

16. 修改 web/index.html 数据管理 Tab：

    • 保留现有状态卡片
    • 新增工具栏：全量更新、增量更新按钮 + since 日期输入（增量可选）
    • 新增任务列表表格：ID（截断）、类型、状态徽章、进度条、完成/总数、时间、取消按钮
    • JS：createTask(type) → POST /api/tasks → 启动轮询 pollTasks()（2s 间隔，有 running 任务时持续，无则停止）
    • renderTaskTable(tasks)：渲染表格 + 进度条
    • cancelTask(id) → POST cancel

关键文件

文件	操作	说明
internal/datasource/eastmoney/client.go	修改	throttle 加锁 + 可配置间隔
internal/cache/cache.go	新建	Cache 接口 + Task 模型 + NoopCache
internal/cache/redis.go	新建	Redis 实现
internal/storage/sqlite/store.go	修改	新增 ingest_task 表 + SaveKlinesBatch + LastKlineDates
internal/storage/sqlite/task.go	新建	任务 CRUD
internal/job/store.go	新建	Store 接口
internal/job/ingest.go	新建	候选列表解析 + 增量起点
internal/job/runner.go	新建	worker 池 + collector
internal/job/job.go	新建	Manager + 限速器
internal/controller/task.go	新建	TaskController
internal/controller/api.go	修改	MarketController 注入共享 Store
internal/router/router.go	修改	注册 task 路由
cmd/server/main.go	修改	装配 Redis/Manager/共享Store
web/index.html	修改	数据管理 Tab 增强
web/style.css	修改	进度条 + 徽章样式

并发拉取 + 批量写入流程

Manager.Start(full)
  → 创建 Task，存 SQLite + Redis
  → go runner.run(ctx, task)

runner.run:
  1. resolveCandidates() → 全量列表或已入库 codes
  2. 预加载：skipSet = ListCodes()，lastDateMap = LastKlineDates()
  3. resultCh = make(chan fetchResult, workers*2)
  4. go collect(resultCh)  // 单一 SQLite 写者
  5. for code in candidates:
       sem <- {}  // worker 槽
       go fetchOne(code):  // 并发拉取
         <-gate  // 全局限速令牌
         klines = ds.FetchKline(code)
         resultCh <- {code, klines, err}
         <-sem
  6. wg.Wait(); close(resultCh)
  7. flush 残余 batch
  8. 更新 Task 状态 → SQLite + Redis

collect:  // 单 goroutine
  batch = {}
  for res in resultCh:
    task.Done++
    if err: task.Failed++; failedCodes.append(code)
    else: task.Success++; batch[code] = klines
    if len(batch) >= 20: store.SaveKlinesBatch(batch); batch = {}
    reportProgress(task)  // Redis 每次 + SQLite 每5只

验证方法

1. 编译：go build ./... 每阶段验证
2. 竞态检测：go test -race ./internal/...
3. 端到端：
   • Docker 启动 Redis
   • 启动 server，admin 登录
   • 数据管理 Tab → 点"增量更新" → 任务表出现 running 行 → 进度条实时增长 → 完成
   • 点"全量更新" → 两个任务并行
   • 点"取消" → 状态变 cancelled
   • 重启 server → 任务列表仍在（SQLite 持久化）
   • 停 Redis → 任务仍能跑（NoopCache 降级）
4. 性能：50 只转债（limit=50）应在 1 分钟内完成