WarrenZhou
273fc90a95
新增完整的异步任务管理体系： 1. 新增Redis缓存层实现任务状态实时存储与降级机制 2. 实现并发拉取+批量写入的高性能入库流程 3. 添加任务管理HTTP接口与路由 4. 重构行情控制器使用共享SQLite存储 5. 配套添加Docker Compose Redis启动配置 6. 前端新增任务管理面板支持操作与进度查看
171 lines
9.0 KiB
Markdown
171 lines
9.0 KiB
Markdown
# 异步任务系统 + 并发拉取批量写入
|
||
|
||
## 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 分钟内完成
|