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
|