hyapp-server/docs/定时任务服务架构.md

14 KiB
Raw Permalink Blame History

Cron Service Architecture And Task Inventory

本文定义 cron-service 的服务边界、运行模型、当前代码中可迁移的后台任务,以及哪些任务不能放进 cron-service。

核心结论:

  • cron-service 是调度和批处理触发进程,不是新的业务状态 owner。
  • 它不能直接 import 其他服务的 internal 包,也不能绕过 owner service 直接改业务表。
  • 房间状态、钱包账务、用户身份、活动消息等业务状态仍由各自服务处理cron-service 只调用 owner service 暴露的内部 gRPC 批处理接口。
  • 不要把所有 time.NewTicker 都迁到 cron-service。事件投递、Room Cell 清理和高频 outbox 消费不是通用 cron 问题。

Current Worker Inventory

当前仓库已经有多类后台 worker

Service Current Worker Current Location Decision
room-service stale presence cleanup services/room-service/internal/room/service/presence.go 留在 room-service
room-service mic publish timeout cleanup services/room-service/internal/room/service/mic_publish_timeout.go 留在 room-service
room-service room outbox relay to direct publishers / RocketMQ services/room-service/internal/room/service/outbox_worker.go 留在 room-service 或后续拆 room-worker,不进通用 cron
room-service room rocket delayed launch wakeup services/room-service/internal/room/service/room_rocket.goservices/room-service/internal/room/service/room_rocket_mq.go 留在 room-serviceRocketMQ 只唤醒,到点后仍走 Room Cell 命令链路
user-service region rebuild task UserCronService.ProcessRegionRebuildBatch 已迁移到 cron-service 调度,旧内置 ticker 已删除
user-service login IP risk job UserCronService.ProcessLoginIPRiskBatch 已迁移到 cron-service 调度,旧内置 ticker 已删除
user-service invite recharge wallet outbox consumer services/user-service/internal/service/invite/service.go 先不迁,需补 durable cursor 或事件总线
user-service room mic outbox consumer services/user-service/internal/service/mictime/service.go 先不迁,属于事件消费,不是普通 cron
user-service mic open session compensation UserCronService.CompensateMicOpenSessions 已迁移到 cron-service 调度,旧内置 ticker 已删除
activity-service message fanout job ActivityCronService.ProcessMessageFanoutBatch 已迁移到 cron-service 调度,旧内置 ticker 已删除
server/admin admin export jobs server/admin/internal/job/runner.go 留在 admin module
wallet-service wallet outbox table exists, no worker services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql 后续做 wallet outbox worker不进通用 cron

Service Boundary

cron-service 的职责是“按计划触发 owner service 处理一批任务”,不是“拿到所有数据库连接后自己处理业务”。

graph LR
  Cron["cron-service\nscheduler + run history"] --> CronDB[("hyapp_cron")]

  Cron -->|"ProcessLoginIPRiskBatch"| User["user-service"]
  Cron -->|"ProcessRegionRebuildBatch"| User
  Cron -->|"CompensateMicOpenSessions"| User
  Cron -->|"ProcessMessageFanoutBatch"| Activity["activity-service"]
  Cron -->|"ExpireVipBatch / SettlementBatch"| Wallet["wallet-service"]

  User --> UserDB[("hyapp_user")]
  Activity --> ActivityDB[("hyapp_activity")]
  Wallet --> WalletDB[("hyapp_wallet")]
  Room["room-service"] --> RoomDB[("hyapp_room")]

边界规则:

Rule Explanation
只调 gRPC cron-service 通过 hyapp.local/api 生成 client 调用内部 RPC不 import 其他服务实现
不直接写业务库 cron-service 只写自己的 run history 和 schedule lock业务表由 owner service 写
owner service 自己 claim 每个批处理 RPC 在 owner service 内完成任务 claim、幂等、事务和状态推进
request meta 仍要透传 cron 调内部 RPC 也必须带 RequestMeta.request_id/app_code/caller
多 App 按 app_code 执行 cron 配置任务的 app_codes,每个 app_code 独立触发,避免跨租户混跑
失败不影响主链路 cron 挂了只造成后台积压,不能阻塞登录、送礼、进房等在线路径

What Goes Into Cron-Service

适合放进 cron-service 的任务必须同时满足:

  • 不是用户请求主链路。
  • 有持久任务表、状态机或可重试条件。
  • 单批处理可幂等,重复调用不会重复入账、重复发奖励或重复清状态。
  • owner service 已经有或可以新增 Process...Batch 这类内部 RPC。
  • 任务失败后可以下一轮重试,且需要运行记录、告警和统一调度。

当前已放入 cron-service 调度的任务:

Priority Task Owner Why
P0 login IP risk job user-service 登录后异步 IP 查询,已有 login_ip_risk_jobs、claim TTL、session revoke 逻辑cron 只触发 ProcessLoginIPRiskBatch
P1 user region rebuild user-service 已有 user_region_rebuild_tasksProcessNextRegionRebuildTaskcron 按 app_code 触发单批
P1 message fanout job activity-service 已有 message_fanout_jobs、cursor、lock TTL、retrycron 触发后 owner service 自己 claim
P2 mic open session compensation user-service 低频补偿任务,修正 user mic facts不反写 Room Cell窗口参数放在 cron task 配置

后续推荐接入 cron-service 的任务:

Priority Task Owner Why
P2 VIP expiration scan wallet-service VIP 文档已定义过期扫描,未来由 wallet-service 批量过期并写 outbox
P2 daily task claim retry / expiry activity-service 活动任务中的 pending claim、过期状态适合后台补偿
P3 host salary settlement trigger user-service + wallet-service 日结、周结、半月结、月结是明确的定时结算入口,但实际工资单和入账仍由 owner service 完成

What Must Not Go Into Cron-Service

以下任务不能放到通用 cron-service

Task Reason
Room Cell stale presence cleanup 依赖本节点已装载 Room Cell 和 Redis lease必须由 room-service owner 执行命令链路
Mic publish timeout cleanup 必须对当前 Room Cell 状态做二次校验并执行 MicDown,不能让 cron 直接清麦
room outbox relay 这是 room-service 的事件投递补偿,不是通用定时任务;需要和 direct publisher、RocketMQ 发布、outbox 状态语义绑定
room rocket launch timer 火箭状态属于 Room Cell延迟消息只负责唤醒发射必须由 room-service 重新校验 rocket_id/launch_at_ms/UTC reset 后执行命令
wallet ledger transaction 金币、钻石、积分、余额的加减和冻结必须在 wallet-service 事务内完成
wallet outbox publish 属于 wallet-service outbox relay 或 MQ 投递,不应该由通用 cron 直接扫表改状态
invite recharge outbox consumer 当前使用 in-memory cursor 扫 wallet outbox迁移前必须先做 durable offset 或正式事件总线
room mic outbox consumer 属于高频事件消费,当前靠消费幂等表兜底;迁移前必须先做 durable offset不能用 cron 从头扫
admin export jobs 属于 server/admin 独立 module涉及后台权限、导出文件和 artifact 下载,继续留在 admin runner

Runtime Model

cron-service 是长运行进程,不是一次性脚本。它本身只需要自己的库和内部服务地址。

建议本地端口:

Port Usage
13007 cron-service gRPC health 和内部管理接口

建议目录:

services/cron-service/
  cmd/server/main.go
  configs/
    config.yaml
    config.docker.yaml
    config.tencent.example.yaml
  deploy/mysql/initdb/
    001_cron_service.sql
  internal/
    app/
    config/
    scheduler/
    storage/mysql/
    integration/
    transport/grpc/

建议配置:

service_name: cron-service
node_id: cron-local
environment: local
grpc_addr: ":13007"
mysql_dsn: "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_cron?parseTime=true&charset=utf8mb4&loc=UTC"

user_service_addr: "127.0.0.1:13005"
activity_service_addr: "127.0.0.1:13006"
wallet_service_addr: "127.0.0.1:13004"

tasks:
  login_ip_risk:
    enabled: true
    interval: "5s"
    timeout: "3s"
    lock_ttl: "30s"
    batch_size: 100
    app_codes: ["lalu"]
  user_region_rebuild:
    enabled: true
    interval: "5s"
    timeout: "5s"
    lock_ttl: "30s"
    batch_size: 500
    app_codes: ["lalu"]
  message_fanout:
    enabled: true
    interval: "1s"
    timeout: "10s"
    lock_ttl: "30s"
    batch_size: 500
    app_codes: ["lalu"]
  mic_open_session_compensation:
    enabled: true
    interval: "60s"
    timeout: "10s"
    lock_ttl: "60s"
    batch_size: 100
    pending_publish_max_age: "2m"
    publishing_session_max_age: "12h"
    app_codes: ["lalu"]

Cron Database

cron-service 需要自己的运行记录和调度锁,避免多个 cron 实例同时触发同一个计划。

CREATE TABLE cron_task_leases (
  task_name VARCHAR(96) NOT NULL,
  app_code VARCHAR(32) NOT NULL,
  owner_node_id VARCHAR(128) NOT NULL,
  lease_until_ms BIGINT NOT NULL,
  updated_at_ms BIGINT NOT NULL,
  PRIMARY KEY (task_name, app_code),
  KEY idx_cron_task_leases_expire (lease_until_ms)
);

CREATE TABLE cron_task_runs (
  run_id VARCHAR(96) NOT NULL,
  task_name VARCHAR(96) NOT NULL,
  app_code VARCHAR(32) NOT NULL,
  owner_node_id VARCHAR(128) NOT NULL,
  status VARCHAR(32) NOT NULL,
  claimed_count INT NOT NULL DEFAULT 0,
  processed_count INT NOT NULL DEFAULT 0,
  success_count INT NOT NULL DEFAULT 0,
  failure_count INT NOT NULL DEFAULT 0,
  started_at_ms BIGINT NOT NULL,
  finished_at_ms BIGINT NULL,
  duration_ms BIGINT NOT NULL DEFAULT 0,
  error_message VARCHAR(512) NOT NULL DEFAULT '',
  PRIMARY KEY (run_id),
  KEY idx_cron_task_runs_task_time (task_name, app_code, started_at_ms),
  KEY idx_cron_task_runs_status (status, started_at_ms)
);

cron_task_leases 只防止调度层重复触发;真正的业务并发安全仍必须由 owner service 的任务表锁和幂等保证。

Internal RPC Contract

不要设计一个可以执行任意任务名的通用 Execute(task_name, payload)。这种接口最后会变成绕过领域边界的后门。

每个 owner service 暴露明确的内部批处理 RPC例如

message CronBatchRequest {
  RequestMeta meta = 1;
  string run_id = 2;
  string worker_id = 3;
  int32 batch_size = 4;
  int64 lock_ttl_ms = 5;
  int64 pending_publish_max_age_ms = 6;
  int64 publishing_session_max_age_ms = 7;
}

message CronBatchResponse {
  int32 claimed_count = 1;
  int32 processed_count = 2;
  int32 success_count = 3;
  int32 failure_count = 4;
  bool has_more = 5;
}

owner service 侧实现要求:

  • RPC 方法只处理一个明确任务,例如 ProcessLoginIPRiskBatch
  • 任务 claim、事务、状态推进、幂等和业务错误都留在 owner service。
  • 返回 has_more=truecron-service 可以在同一轮继续拉下一批,直到空批次或达到单轮最大耗时。
  • RPC 必须能安全重复调用。cron 超时后无法确认结果时,下一轮重试不能造成重复副作用。

Current Implementation

当前实现已经完成以下迁移:

  1. login_ip_risk_jobscron-service 调 user-serviceProcessLoginIPRiskBatch,旧 RunLoginIPRiskWorker 已删除。
  2. user_region_rebuild_taskscron-service 调 user-serviceProcessRegionRebuildBatch,旧 RunRegionRebuildWorker 已删除。
  3. message_fanout_jobscron-service 调 activity-serviceProcessMessageFanoutBatch,旧 RunFanoutWorker 已删除。
  4. user_mic_sessions open session compensationcron-service 调 user-serviceCompensateMicOpenSessions,旧 RunOpenSessionCompensationWorker 已删除。

cron-service 可以直接开启多个实例;调度层通过 cron_task_leases 防重复触发,业务层仍通过 owner service 内部的任务 claim、状态机和幂等保证最终正确。

Event Consumers Stay Separate

以下不是 cron-service 的首要工作:

  • invite recharge wallet outbox consumer。
  • room mic outbox consumer。
  • wallet outbox relay。
  • activity room outbox consumer。
  • notice room outbox consumer。
  • room IM bridge MQ consumer。
  • room rocket delayed launch consumer。

这些需要的是事件消费基础设施RocketMQ、Redis Stream、或 MySQL durable offset。cron-service 不能用“每隔几秒从头扫 outbox”的方式替代事件消费者。即使某些消费者内部有低频扫描补偿也必须留在业务 owner service 中,因为它们需要理解 outbox 状态、消费幂等和领域状态校验。

Future Scheduled Business Jobs

在对应业务落地后再接入:

  • VIP 过期扫描。
  • 主播工资周期结算触发。
  • 钻石兑换过期订单补偿。
  • 提现人工打款超时提醒。
  • daily task 过期和 pending reward claim 重试。

Operational Notes

Area Requirement
Shutdown 停止调度新 run等待当前 RPC 在 timeout 内结束,再关闭连接
Logs 固定输出 task_name/run_id/app_code/worker_id/duration_ms/claimed_count/processed_count/status
Metrics 每个任务记录积压、成功数、失败数、最老 pending age、运行耗时
Alerts 连续失败、积压持续增长、最老 pending 超阈值、owner service gRPC unavailable
Backoff 单任务连续失败时指数退避,但不能影响其他任务
Idempotency cron-service 只负责少触发,业务不重复必须由 owner service 保证

Final Placement Decision

当前 cron-service 已经承接四个成熟批处理任务;后续只继续接明确的定时批处理,不接在线主链路和事件消费者。

最终放置原则:

  1. login_ip_risk_jobsuser_region_rebuild_tasksmessage_fanout_jobs、mic open session compensation 由 cron-service 调度。
  2. owner service 暴露明确 RPCcron-service 不直接访问业务库。
  3. outbox/event consumer 单独做 durable offset 或 MQ不混进 cron-service。
  4. 未来 VIP、主播工资、daily task 这类周期任务统一接入 cron-service。