chakcy
f4de43c743
单任务受控执行（15 态状态机 + 审计轨迹）、DAG 管道、Controller 门面。
执行器三个：线程（默认，含 async def 任务）、内联（同步 / 测试）、进程
（spawn 子进程：硬性停止、绕开 GIL、崩溃隔离）。
调度：并发额度 + 每桶限流 + 具名资源信号量（按标签）+ 优先级排队。
持久层：TaskStore 契约 + 内存 / SQLite / Postgres 实现，契约测试随包发布。
跨进程与多节点：写穿 / 读穿、指令通道与租约、owned_buckets 划界、reconcile。
CLI、12 个可直接运行的示例、docs/ 下 5 份文档。
门禁：ruff / ruff format / mypy(strict) 干净；
pytest 563 passed（44 skipped 是需要真实 Postgres 的那批）；
带一台可连的 Postgres 时 607 passed。
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""一个可以直接用的常驻任务模块（CLI 的 ``--tasks`` 就指向它）。
|
|
|
|
它是给命令行用的示例模块，也是你自己模块的模板。用法::
|
|
|
|
PYTHONPATH=examples uv run chongming-task --db ./tasks.db \
|
|
run --tasks worker_tasks consume --arg source=kafka://orders --task-id orders-worker
|
|
|
|
## 三件必须做对的事
|
|
|
|
1. **每轮调 ``ctx.checkpoint()``**——停止是协作式的，不调用它就停不掉（会落 ``lost``）。
|
|
2. **用 ``ctx.is_stop_requested`` 收尾**——这样任务终态是 ``succeeded``，返回值保留；
|
|
让 ``checkpoint()`` 抛则是 ``stopped``、无返回值。
|
|
3. **进度按批次上报，不要每条一次**——``ctx.progress()`` 每次都要取运行锁，它是给
|
|
"人看得懂"的频率设计的。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
from chongming_task import RetryPolicy, task
|
|
|
|
|
|
@task(tags={"kafka"}, retry=RetryPolicy(max_attempts=1))
|
|
def consume(ctx: Any, source: str, batch: int = 1000) -> int:
|
|
"""常驻消费 source；收到停止请求后收尾并返回本次共处理多少条。"""
|
|
ctx.log(f"开始消费 {source}")
|
|
total = 0
|
|
while not ctx.is_stop_requested:
|
|
ctx.checkpoint()
|
|
|
|
# 假装取了一批消息。真实实现里这里是一次 poll / 一次拉取。
|
|
time.sleep(0.05)
|
|
total += batch
|
|
|
|
# 按批次上报进度（不是每条一次）。
|
|
ctx.progress(total, unit="msg", message=source)
|
|
|
|
ctx.log(f"收到停止请求，收尾：本次共处理 {total} 条")
|
|
return total
|
|
|
|
|
|
@task
|
|
def once(ctx: Any, value: int) -> int:
|
|
"""一个短任务，用来对照"有始有终"的普通形态。"""
|
|
del ctx
|
|
return value * 2
|