csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity
Files
e77bb0d09b4da61e5dbe787080714dcafdbe73bb
junhong_cmp_fiber/openspec/changes/archive/2026-04-07-polling-system-refactor/proposal.md
huang 434a8b0349
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m46s
Details
feat: 轮询系统重构（分片队列 + 停复机统一 + Handler 拆分） 
【核心变更】

1. 停复机逻辑统一（StopResumeService）
   - 新增 EvaluateAndAct 统一入口，封装三条件停复机判断
   - 停机条件：无套餐(no_package) / 流量耗尽(traffic_exhausted) / 未实名(not_realname)
   - 复机条件：stop_reason 合规 + 有套餐且未耗尽 + 已实名或行业卡
   - 修复设备套餐 Bug：hasValidPackage 按 device_id 查套餐，而非仅 iot_card_id
   - 设备维度停复机加幂等锁（Redis SetNX，TTL 30s），防止多卡并发重复调 Gateway

2. Redis 分片队列（PollingQueueManager）
   - 新建 queue_manager.go，封装所有轮询 Redis 操作
   - 16 分片 Sorted Set，Key 格式：polling:shard:{shardID}:queue:{taskType}
   - Lua 脚本原子出队（ZRANGEBYSCORE + 分批 ZREM），消除竞态窗口
   - 新增背压检测：队列深度超 50 万时 Scheduler 跳过该分片
   - RemoveFromAllQueues 覆盖 4 种任务类型（含 protect）

3. Handler 拆分（polling_handler.go 1360行 → 5个专注文件）
   - polling_base.go：共享基类（并发控制/卡缓存/重入队）
   - polling_realname_handler.go：实名采集，实名 0→1 时立即触发复机
   - polling_carddata_handler.go：流量采集，保留跨月边界检测逻辑
   - polling_package_handler.go：套餐采集，委托 EvaluateAndAct 决策
   - polling_protect_handler.go：保护期一致性检查，保护期内强制修正

4. 配置管理（PollingConfigManager）
   - 新建 config_manager.go，从 scheduler.go 提取配置职责
   - 内存缓存 + 5 分钟定时刷新，刷新失败保留原缓存
   - 修复 getCardCondition：停机卡返回 suspended，不再错配 activated 配置

5. 渐进式初始化（CardInitializer）
   - 新建 initializer.go，分批加载（每批 10 万），批次间 sleep 500ms
   - 过滤 enable_polling=false 的卡，初始化完成前 Scheduler 不出队

6. 卡生命周期服务（PollingLifecycleService）
   - 新建 lifecycle_service.go，替代已删除的 callbacks.go 和 api_callback.go
   - OnCardCreated/OnCardEnabled/OnCardStatusChanged 入队前检查 enable_polling

7. Scheduler 精简（1000+行 → 227行）
   - 保留纯调度循环：scheduleLoop + processShardSchedule + enqueueBatch
   - 保留每 10 秒触发套餐过期检测和流量重置
   - 移除所有 DB 操作、配置加载、卡初始化逻辑

8. 轮询管控 API（enable_polling）
   - 新增 PUT /api/admin/assets/:id/polling-status 接口
   - 支持对设备/卡维度开关轮询，关闭后从所有分片队列移除

9. 数据库迁移
   - 000103：tb_device 新增 enable_polling 字段（boolean, NOT NULL, DEFAULT true）
   - 000104：新增 suspended 轮询配置，为 activated 配置补全 protect_check_interval

【文件统计】
- 新增：19 个文件（handler × 5、polling 组件 × 4、迁移 × 3 等）
- 修改：20 个文件（bootstrap 注入、store 接口、monitoring 适配分片等）
- 删除：3 个文件（polling_handler.go、callbacks.go、api_callback.go）

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
2026-04-07 12:27:04 +08:00

11 KiB
Raw Blame History

Why（已合并 polling-logic-redesign，可废弃该提案）

轮询系统经过多次迭代后出现了四类问题，须同步解决：

一、结构性问题（影响长期可维护性）

问题 1：polling_handler.go（1360行）职责爆炸 Asynq Task Handler 本应只做「数据采集」，但它同时承担：停复机决策（与 stop_resume_service.go 重复）、直接 DB 操作（15处绕过 Store 层）、直接 Gateway 调用（7处无重试）、缓存管理、队列管理。

问题 2：停复机逻辑分裂，可靠性不一致 polling_handler.go 和 stop_resume_service.go 各有一套停复机实现，hasAvailablePackage 重复定义；前者 Gateway 调用无重试，后者有 3 次重试，同一业务，不同可靠性。

问题 3：Scheduler（764行）身兼六职 调度循环 + 配置管理 + 卡初始化 + 生命周期回调 + 套餐激活触发 + 流量重置触发。callbacks.go 与 api_callback.go 大量重复代码（两套相同的 Redis 队列操作逻辑）。

二、线上 Bug（需立即修复）

Bug 1（停复机判断错误）：设备套餐被忽略 → 误停机 PackageUsage 设备套餐存 device_id，但 shouldStopCard / hasAvailablePackage 只查 iot_card_id，购买设备套餐后卡仍被误停机。

Bug 2（停复机条件不完整）：停机条件缺失 当前停机条件仅为「无有效套餐」，未判断虚流量是否耗尽，未区分行业卡/非行业卡的实名要求。

Bug 3（保护期队列遗漏）：protect 队列残留脏数据 removeFromAllQueues 只清 3 个队列（realname、carddata、package），漏掉 protect 队列，删除卡后 protect 队列残留。

Bug 4（出队竞态）：ZRANGEBYSCORE + ZREMRANGEBYSCORE 非原子 两层风险：① 高并发下多 Worker 读取同一批卡，导致重复停机/复机 Gateway 调用；② ZRANGEBYSCORE 有 LIMIT 但 ZREMRANGEBYSCORE 无 LIMIT，当到期卡数超过批次大小时，超出部分被删除但从未被读取——卡永久丢失，不再被轮询。

三、规模挑战（千万级支撑设计）

当前设计假设百万级（100K 初始化批次、50K 调度批次、50 并发）。千万级规模下，单个 Sorted Set 深度达千万条，需引入：分片队列（Sharded Sorted Set）、多 Worker 分片消费（无协调开销）、分片并行初始化。

四、管理接口兼容性确认

现有 36 个轮询管理 HTTP 接口（配置/手动触发/告警/监控/并发/清理）经全量检查，绝大部分 Service 只依赖 Redis Client 和 Store，不依赖 Scheduler 对象。其中 32 个接口全部兼容，零改动；4 个监控统计接口需适配分片队列（MonitoringService 读取队列深度的 Key 从 polling:queue:{type} 变更为聚合所有分片 polling:shard:{N}:queue:{type} 的 ZCard 之和）。

What Changes

功能修复（来自 polling-logic-redesign，合并入本提案）：

  • 修复 hasAvailablePackage / shouldStopCard：根据卡绑定关系动态切换 iot_card_id / device_id 查询
  • 重写停复机判断：新增三条件停机（无套餐/流量耗尽/未实名），区分行业卡
  • 新增停机原因精细化常量：no_package、not_realname
  • 新增 Device.enable_polling 字段
  • 新增 HTTP 接口：PATCH /api/admin/assets/:asset_type/:id/polling-status

架构重构：

  • 拆分 polling_handler.go（1360行）→ 4 个专注 Handler（实名/流量/套餐/保护期）+ 共享基类，每个 < 300行
  • 新增 PollingQueueManager：统一 Redis 队列操作，Lua 脚本原子出队（ZRANGEBYSCORE + ZREM 原子执行），修复 protect 遗漏 Bug
  • 新增 PollingConfigManager：独立配置管理，支持 5 分钟定时刷新
  • 新增 CardInitializer：独立初始化模块，支持分片并行
  • 精简 Scheduler（目标 < 250行）：保留调度循环 + 套餐过期检测触发 + 流量重置触发（PackageActivationHandler.HandlePackageActivationCheck 和 DataResetHandler.HandleDataReset 必须保留在调度主循环中）
  • 新增 PollingLifecycleService：替代 callbacks.go + api_callback.go 中卡生命周期方法（OnCardCreated/OnCardStatusChanged/OnCardEnabled/OnCardDisabled/OnCardDeleted），依赖 PollingQueueManager + PollingConfigManager，封装「匹配配置 + 入队」组合逻辑
  • 删除 callbacks.go + api_callback.go（队列操作由 PollingQueueManager 替代，生命周期方法由 PollingLifecycleService 替代）
  • MonitoringService 适配分片队列：队列深度查询改为聚合所有分片的 ZCard 之和

千万级规模设计：

  • 分片 Sorted Set：polling:shard:{0..N-1}:queue:{taskType}，默认 16 分片
  • CardInitializer 按 card_id % shard_count 分桶入队
  • Scheduler 并行消费 N 个分片（每个分片独立 Lua 脚本原子出队）
  • 背压机制：单分片队列深度超阈值时，跳过该分片本轮调度

重要设计约束：

  • Asynq 任务提交保持 MaxRetry(0)（与当前设计一致），失败后通过 requeueCard 放回 Redis Sorted Set 延后处理，避免 Asynq 重试与 Scheduler 出队产生并发双重处理
  • StopResumeCallback 接口不在本次变更范围内（仅替换 PollingCallback），triggerStopAfterExpiry() 和 checkAndTriggerSuspension() 不受影响
  • Phase 4（Task Handler 拆分）和 Phase 5（Scheduler 精简）必须原子部署，不可分开上线（详见迁移计划）

Capabilities

New Capabilities

  • polling-queue-manager: 统一 Redis 队列操作封装——Lua 脚本原子出队（ZRANGEBYSCORE + ZREM 服务端原子执行，保留时间过滤语义）、分片 Sorted Set（支持千万级）、修复 protect 遗漏 Bug、入队/重入队/手动触发/删除的统一接口；替代 callbacks.go、api_callback.go、polling_handler.go 中所有分散的队列操作
  • polling-config-manager: 独立配置管理模块——从 DB 加载 tb_polling_config、同步到 Redis Hash（TTL 24h）、内存缓存（读写锁）、5 分钟定时自动刷新、MatchConfig(card) 按优先级返回第一个匹配配置
  • card-initializer: 独立卡初始化模块——渐进式分批（100K/批）、按分片入队（card_id % shard_count）、enable_polling=false 过滤、进度状态暴露给 MonitoringService
  • polling-lifecycle-service: 卡生命周期轮询管理——替代 callbacks.go 和 api_callback.go 中的 OnCardCreated/OnBatchCardsCreated/OnCardStatusChanged/OnCardEnabled/OnCardDisabled/OnCardDeleted，依赖 PollingQueueManager（队列操作）+ PollingConfigManager（配置匹配），封装「匹配配置 → 分片入队」组合逻辑；两个进程（API 和 Worker）共享
  • asset-polling-control: 资产轮询管控 HTTP 接口——PATCH /api/admin/assets/:asset_type/:id/polling-status，支持 card/device 两种资产类型，启用/禁用轮询

Modified Capabilities

  • polling-stop-resume-logic: 停复机逻辑统一到 StopResumeService——新增 EvaluateAndAct() 统一入口、修复设备套餐查询 Bug（device_id vs iot_card_id）、实现三条件停机判断（无套餐/流量耗尽/未实名）、完整复机条件（含行业卡豁免）、停机原因精细化
  • polling-task-handlers: polling_handler.go 拆分为 4 个专注文件——每个 Handler 只做数据采集，停复机通过 StopResumeService.EvaluateAndAct() 完成，消除所有直接 DB 操作和无重试 Gateway 调用；polling_carddata_handler.go 须完整保留当前 HandleCarddataCheck 中的跨月流量边界检测逻辑（月份切换检测、上月总量保存、当月计数器重置）
  • polling-monitoring-service: MonitoringService 适配分片队列——队列深度查询改为调用 PollingQueueManager.GetTotalQueueDepth(taskType) 聚合所有分片，替代直接读取旧的非分片 Redis Key

Impact

新建文件

文件 职责 目标行数
internal/task/polling_realname_handler.go 实名检查 Task Handler < 200行
internal/task/polling_carddata_handler.go 流量检查 Task Handler < 300行
internal/task/polling_package_handler.go 套餐检查 Task Handler < 200行
internal/task/polling_protect_handler.go 保护期一致性 Task Handler < 200行
internal/task/polling_base.go 共享基类（并发控制/缓存/重入队） < 150行
internal/polling/queue_manager.go 统一 Redis 队列操作 < 200行
internal/polling/config_manager.go 配置加载管理 < 150行
internal/polling/initializer.go 分片渐进式初始化 < 250行
internal/polling/lifecycle_service.go 卡生命周期轮询管理（替代 callbacks.go） < 200行

修改文件

文件 变更类型 变更内容
internal/polling/scheduler.go 精简重写 保留调度循环 + 套餐过期/流量重置触发，< 250行
internal/service/iot_card/stop_resume_service.go 扩展+修复 新增 EvaluateAndAct + 修复设备套餐 Bug + 三条件停机
internal/model/device.go 字段新增 EnablePolling bool
internal/handler/admin/asset.go 方法新增 UpdatePollingStatus handler
internal/routes/asset.go 路由新增 PATCH /assets/:asset_type/:id/polling-status
internal/store/postgres/device_store.go 方法新增 UpdatePollingStatus(ctx, id, enabled)
pkg/constants/iot.go 常量新增 StopReasonNoPackage、StopReasonNotRealname
pkg/constants/redis.go 函数新增 RedisPollingShardQueueKey(shard, taskType)
internal/service/polling/monitoring_service.go 适配更新 队列深度查询改为聚合分片 ZCard
cmd/worker/main.go 启动流程更新 使用新组件，注册 4 个 Handler
pkg/queue/handler.go Handler 注册更新 注册 4 个新 Task Handler
cmd/api/docs.go + cmd/gendocs/main.go 文档更新 注册新 AssetHandler 方法

删除文件

文件 删除原因
internal/task/polling_handler.go 拆分为 4 个专注文件后废弃
internal/polling/callbacks.go 职责转移到 PollingQueueManager
internal/polling/api_callback.go 职责转移到 PollingQueueManager

数据库迁移

表 变更
tb_device 新增 enable_polling BOOLEAN NOT NULL DEFAULT TRUE

管理接口兼容性（零改动）

模块 接口数 兼容性
轮询配置管理（polling-configs） 7个 ✅ 完全兼容
手动触发（polling-manual-trigger） 6个 ✅ 完全兼容
告警管理（polling-alert-rules） 6个 ✅ 完全兼容
监控统计（polling-stats） 4个 ⚠️ 需适配分片队列（聚合 ZCard）
并发控制（polling-concurrency） 4个 ✅ 完全兼容
数据清理（data-cleanup） 9个 ✅ 完全兼容
合计 36个 32个零改动 + 4个监控接口需适配
Reference in New Issue View Git Blame Copy Permalink