chongming-utils-task/docs/best-practices.md

最佳实践

这份文档只回答一个问题：照什么做，生产上不会踩坑。

它不是入门教程（那是 用户指南），也不是接口规格（那是 持久层实现指南 与 设计文档）。这里给的是 默认答案——绝大多数场景下该选什么、任务该怎么写、哪些坑已经有人踩过了。

一条贯穿全文的原则：这个库是 at-least-once 的。停止、超时、进程崩溃之后，任务 可能已经产生过一部分副作用。所有"最佳实践"最终都落回同一句话——把任务写成幂等的。

---

0. 一分钟版

from chongming_task import (
    BucketPolicy, ConcurrencyPolicy, Controller, ExponentialBackoff, RetryPolicy,
    SqliteTaskStore, TaskContext, task,
)

NUM_BUCKETS = 64          # 部署级约定：所有节点必须一致


@task(tags={"sync", "db"}, retry=RetryPolicy(max_attempts=3, backoff=ExponentialBackoff()))
def sync_tenant(ctx: TaskContext, tenant: str) -> dict[str, object]:
    """同步一个租户的数据。"""
    pages = list_pages(tenant)                            # 你的业务代码
    for index, page in enumerate(pages, start=1):
        ctx.checkpoint()                                  # ← 必须
        ctx.progress(index, total=len(pages), message=tenant)  # 进度走旁路，不碰版本号
        apply_page(tenant, page)                          # ← 必须幂等
    return {"tenant": tenant, "pages": len(pages)}


controller = Controller(
    store=SqliteTaskStore("tasks.db"),                    # 跨机换成 PostgresTaskStore(dsn)
    # owner 默认就是 f"{主机名}:{进程号}"，要区分同一台机上的多个节点再显式给。
    bucket_policy=BucketPolicy(num_buckets=NUM_BUCKETS),
    owned_buckets=range(0, 32),                           # 本节点负责的桶（连续分区）
    concurrency=ConcurrencyPolicy(max_parallel=8, limits={"db": 4}),
    max_parallel_per_bucket=1,                            # 同一租户串行
)
controller.register_all(sync_tenant)
controller.reconcile()                                    # 启动时清理上次的僵尸任务

五条不可省略的：

必须	为什么
循环里调 ctx.checkpoint()	不调 = 停不掉、也会被 idle 判成卡死
任务幂等	超时/停止/崩溃之后可能重跑，副作用会再来一次
返回值可 JSON 序列化（跨进程再加"可 pickle"）	返回值要落库；不满足会在返回时响亮失败
进程执行器的目标函数在模块里	spawn 的子进程靠"模块 + 函数名"找回它
NUM_BUCKETS 各节点一致	不一致会让同一个键落到不同的桶，且不报错

---

1. 按规模选型

场景	存储	执行器	起点
脚本里跑一个任务	不要存储	线程（默认）	run_task(fn, *args)
单机常驻服务	SqliteTaskStore("tasks.db")	线程	本文 §0
多机 / 多进程共用一套状态	PostgresTaskStore(dsn)	线程（按需进程）	示例 11

内存存储只用于测试。 InMemoryTaskStore 不共享、不持久化——用它跑生产等于把 "跨进程可见"和"重启恢复"这两件事一起放弃。

能单机就别多机。 多节点带来的是运维成本（桶划分、告警巡检），不是性能魔法； 单机的 max_parallel 通常先到瓶颈，先把它调明白。

---

2. 任务怎么写

2.1 签名与位置

@task(tags={"db"})
def sync_tenant(ctx: TaskContext, tenant: str, *, batches: int = 24) -> dict[str, object]:
    ...

ctx 由执行器注入为第一个位置参数，提交时不要自己传。需要跨进程执行时，任务函数 必须是模块级的（不能是闭包、lambda、局部函数）——这一点在进程执行器下是硬约束， 违反时提交那一刻就报错。

2.2 检查点频率要与 idle 对齐

idle 超时衡量的是两次 checkpoint() 之间的间隔，不是"CPU 有没有在干活"。

controller.submit("calc", timeout=Timeout(idle=30))     # 30 秒没有检查点 = 判卡死

• 任务每批最多 2 秒 → idle=30 很宽裕；
• 任务有一段"一口气算 5 分钟"的循环 → 要么在里面塞 checkpoint，要么把 idle 调到 6 分钟以上。漏了这一步的症状是：任务跑得好好的，被判 timeout_idle。

ctx.progress() 不刷新心跳。 这是刻意的：进度是"到哪了"，心跳是"还在推进"。 只报进度不调检查点的任务会被如实判成卡死——那正是 idle 要抓的东西。

2.3 幂等：把"重跑一次"当常态

会重跑的情况不止失败重试：

情况	之后会发生什么
stop() 落在任务收尾窗口里	状态 stopped，但副作用可能已发生一半
优雅窗口用尽（线程）	状态 lost，任务可能还在后台跑，随后可能重试
优雅窗口用尽（进程）	子进程被 kill()，已写入的东西不会回滚
进程崩溃 / 被 OOM 杀掉	同上，且状态是 lost（结果未知）

所以：写操作要能重放——INSERT ... ON CONFLICT、先查后写、写临时路径再原子改名、 带幂等键调用外部 API。不要指望"库会保证只跑一次"，它保证不了。

2.4 返回值与参数

@task
def extract(ctx) -> dict[str, object]:
    return {"rows": 3, "path": "s3://bucket/x.parquet"}   # ✅ 传指针
    # return {"frame": df}                                 # ❌ DataFrame 不可 JSON 序列化

返回值会写进存储（outcome 列）：

• 所有执行器：必须可 JSON 序列化，否则任务在返回时失败（result_invalid）；
• 进程执行器：还要能 pickle（先跨进程传回来，再落库）。两个条件都要满足：pathlib.Path 能被 pickle 但不能 JSON 化，所以它做返回值一样会被拒。

参数则相反：线程执行器下随便传（Path、连接对象、句柄都行），进程执行器下必须 可 pickle。所以"参数里塞一个打开的数据库连接"这种写法会锁死执行器的选择。

2.5 别把停止信号吃掉

try:
    while work():
        ctx.checkpoint()
except TaskCancelled:
    raise                     # ← 必须放行：取消/停止是控制流，不是错误
except Exception as exc:      # 业务异常照常处理
    log(exc)

TaskCancelled 继承 BaseException 就是为了防止 except Exception 把它吞掉。它一旦 被吞，任务就永远停不下来。

---

3. 并发与资源

controller = Controller(
    concurrency=ConcurrencyPolicy(
        max_parallel=20,                    # 同时执行上限；超出排队（不是拒绝）
        limits={"db": 4, "gpu": 1},         # 具名信号量：按任务标签记账
        priority=0,                          # 默认优先级
    ),
    max_parallel_per_bucket=1,               # 每桶各自限流
)
controller.submit("sync", "tenant-A", tags=["db"], bucket_key="tenant-A")
controller.submit("report", priority=10)     # 急件先出队

三条经验：

1. 保护外部资源用 limits，不要靠调小 max_parallel。 混合型负载里，为了保住 4 个数据库连接把全局并发压到 4，等于让纯 CPU 的任务跟着一起挨饿。
2. max_parallel_per_bucket=1 是"同一份数据不自我竞争"的最省事做法：同一个租户 （同一个 bucket_key）串行，不同租户仍然并行。
3. 名额与优先级都是本进程的账，不做跨进程协调：多节点各自配自己的份额 （例如"每个节点最多 4 个数据库连接"）。limits 统计的是活跃任务（含被暂停的）， 因为暂停不释放执行槽。

concurrency= 与 max_parallel= 只能给一个（同时给会报错）——两份配置谁说了算 必须明确。

---

4. 执行器怎么选

	线程（默认）	进程
共享内存 / 连接池 / 缓存	✅ 直接复用	❌ 各自一份
闭包、lambda、局部函数	✅	❌ 必须模块级 + 可 pickle
参数的额外要求	无	必须可 pickle
停止 / 暂停	即时	约 20ms 轮询延迟
不响应停止的任务	只能放弃，记 lost（未知）	terminate() → kill()，真的停掉
CPU 密集	受 GIL 限制	真并行
原生扩展段错误	带走整个监督进程	只带走子进程

默认线程。 只在三种场景换 ProcessExecutor()：需要"停得住"的保证（例：任务里跑 别人写的、可能不响应取消的代码）、CPU 密集、或跑第三方原生扩展。

controller = Controller(executor=ProcessExecutor())        # 整批
outcome = run_task(fn, executor=ProcessExecutor())          # 单个

CLI 也有：chongming-task --db tasks.db run --tasks myapp.tasks heavy --executor process。

进程执行器的三条代价（可 pickle、模块级目标 + __main__ 守卫、轮询延迟）见 用户指南 §7.4。注意：别为了 "更安全"无脑上进程——它的约束会渗透到任务签名与参数里，撤回来的成本很高。

---

5. 持久化与重启

5.1 心智模型：内存是缓存，存储是事实

Controller(store=...) 之后：每次状态迁移写穿到存储，查询读穿存储（因此能看到 别的进程的任务），控制动作对不属于本进程的任务指令化。

存储出问题不会让任务失败（可观测性坏了不是任务的错），但也不会静默：

controller.persist_failures     # 写穿失败的历史（去重、有上限）
controller.desynced_tasks       # 检测到"外部写入者"而停止写穿的任务

上线后把这两个属性接进你的告警——它们是"存储里的事实与内存里的状态分叉了"的唯一信号。

5.2 启动时跑一次 reconcile()

report = controller.reconcile()          # 先看一眼：reconcile(dry_run=True)
if report.lost:
    log.warning("上次进程留下的 %d 个任务判为丢失：%s", len(report.lost), report.lost)

判据是监督者租约（"还有人在管它吗"），不是任务心跳——所以长时间不调 checkpoint 的 健康任务不会被误杀。lease_ttl（默认 30 秒）决定崩溃多久之后被发现。

它不会重新执行任何任务。 存储里记的是状态，不是配方：任务的函数与参数从未落库。 重跑必须由调用方重新 submit，而那需要调用方知道参数——这是设计边界，不是遗漏。

5.3 规模上来之后

• prune(keep_terminal=N) 只清内存里的终态记录，存储里的历史仍在；
• transitions 只增不减，长期跑要按月归档或分区（数据库侧的事，库不代劳）；
• 跨机把 SqliteTaskStore("tasks.db") 换成 PostgresTaskStore(dsn) 即可，业务代码 一行不用改；只读运维账号也能连（检测到表已存在就不做 DDL）。

---

6. 多节点与桶划分

桶是节点之间的划分单位：每个节点自治（自己的任务定义与触发方式），但共用一套存储 与控制层，靠 owned_buckets 划界。

def buckets_for_node(index: int, total: int, num_buckets: int) -> range:
    """连续分区——不是取模。取模会让"扩一个节点"把所有桶都重新洗一遍。"""
    per_node = num_buckets // total
    start = index * per_node
    stop = num_buckets if index == total - 1 else start + per_node
    return range(start, stop)

三条铁律：

1. num_buckets 是部署级约定，各节点必须一致（CLI 也要用 --num-buckets 显式给）， 不一致不会报错，只会让任务落到没人拥有的桶里；
2. 提交时校验是保护，不是障碍：落在别人桶里的任务不会有节点执行，它会在 queued 上静默堆积——所以 submit 会直接报错；
3. reconcile() 只看自己的桶，跨节点停止走指令通道（stop() 返回 False 表示 "已下发、未确认"，看结果要查 status）。

没有自动告警：两个节点配了同一个桶（会重复执行）、或某个桶没人管（任务静默堆积）， 都不会报错。这是已知缺口，用你自己的巡检兜住：定期 buckets 视图比对节点配置。

---

7. 运维

7.1 排障三件套

chongming-task --db tasks.db ps --all            # 现在都在什么状态
chongming-task --db tasks.db status <task_id>    # 详情 + 迁移历史（谁在什么时候改的）
chongming-task --db tasks.db buckets             # 哪个桶在堆积（星号是本节点负责的）

任务卡住时先看快照里的 pid：进程执行器下它就是那个子进程的进程号，ps 能直接对上号。

7.2 失败类型决定下一步

failure.kind	含义	建议
exception	任务自己抛异常	看 traceback 修逻辑
timeout_start / timeout_attempt / timeout_idle	三种超时（启动 / 单次尝试 / 静默）	调阈值或加检查点；idle 多半是缺 checkpoint
nonzero_exit	子进程非零退出或崩溃（带 exit_code）	查子进程 stderr；原生扩展崩溃就该用进程执行器隔离
result_invalid	返回值不满足持久化/跨进程契约	改返回值为可序列化/可 pickle，重试无用
cancelled	停止或取消（不是错误）	不用管
lost	结果未知（载体被放弃）	重跑前先人工核对副作用：可能已经生效过一次

7.3 优雅退出

try:
    controller.wait_all(timeout=60)
finally:
    controller.close(timeout=30, close_store=True)   # 关闸 → 停任务 → 补写穿 → 关存储

close() 会先把排队中的任务"启动后立刻停止"（不执行任务体），再等收尾，最后把因瞬时 故障积压的写穿补一次。配 systemd 时用 SIGTERM → 这个路径；别用 kill -9， 那会留下一堆 running 僵尸任务，下次启动靠 reconcile() 收拾。

---

8. 反模式清单

反模式	症状	正确做法
任务里不调 ctx.checkpoint()	停不掉、被判 timeout_idle	每个循环都调；频率与 idle 对齐
用 except Exception 包住循环	停止请求被吞，任务永远停不下来	except TaskCancelled: raise 放行
任务写成非幂等	重跑后数据重复/重复扣款	幂等键、ON CONFLICT、原子替换
返回值塞大对象 / DataFrame	result_invalid 失败	传指针（对象存储 key、表名 + 主键）
参数里传数据库连接	换进程执行器时全部报错	任务内部自建连接池，用 limits 限流
全局并发设为 1 来保护数据库	所有任务串行，吞吐崩掉	用 limits={"db": N} 只限那一类任务
忘了 max_parallel_per_bucket	同一租户的多个任务互相抢锁/打架	同租户串行：max_parallel_per_bucket=1
用内置 hash() 分桶	跨进程桶位不一致，且静默	用 BucketPolicy（blake2b，跨进程稳定）
节点桶数不一致	任务落到没人拥有的桶，静默堆积	num_buckets 当部署配置管，写进巡检
多个进程共用 SQLite 文件	网络盘上锁不可靠	同机用 SQLite 文件、跨机换 Postgres
Controller 用完不 close()	文件句柄泄漏，Windows 下数据库文件删不掉	close(close_store=True)
忘了启动时 reconcile()	看板上永远挂着几个假的"运行中"	启动第一件事就是它
把 lost 当"没跑过"	放心重跑 → 副作用来两次	lost = 结果未知，先核对再决定
无脑上进程执行器	任务签名/参数被 pickle 约束锁死	需要那三种能力时才用（§4）

---

9. 上线检查清单

• 存储选好并接上（生产用 SQLite 文件或 Postgres，不用内存）
• 启动时跑一次 reconcile()，lost 结果进日志/告警
• 退出路径是 close(close_store=True)，systemd 用 SIGTERM
• 每个任务的循环里都有 ctx.checkpoint()，且与 idle 阈值匹配
• 任务幂等（写操作可重放），或明确接受重复副作用
• 返回值可 JSON 序列化（跨进程还需可 pickle）
• max_parallel 定过压测；外部资源用 limits 单独限流
• 需要同租户串行的地方配了 max_parallel_per_bucket
• num_buckets 与各节点的 owned_buckets 对得上（连续分区、无重叠、无空洞）
• persist_failures / desynced_tasks 接进了告警
• 桶重复/空洞有自己的巡检（库不会替你报）
• 执行器选择有理由：默认线程；进程只在需要强杀/GIL/隔离时用

---

相关文档

想要	去哪
完整概念与 API	用户指南
命令行控制台	命令行
换/写持久层	持久层实现指南
多节点架构、边界与风险	分布式设计说明
状态模型、转移表、取舍理由	设计文档
可直接运行的代码	examples/（11 号是集成主模板，12 号讲进程执行器）