Chancy项目中的任务(Jobs)机制详解

什么是Chancy任务

在Chancy项目中，任务(Jobs)是整个系统的核心组件。任务本质上是被工作线程(workers)执行的函数，它们构成了异步任务处理的基础单元。理解并掌握任务的创建和使用方法，是高效利用Chancy框架的关键。

创建任务

基础任务创建

使用@job装饰器可以轻松地将普通函数转换为Chancy任务：

from chancy import job

@job()
def greet():
    print("Hello world!")

这个被装饰的函数仍然可以像普通函数一样调用：

greet()  # 输出: Hello world!

配置任务参数

在创建任务时，可以指定多种参数来定制任务行为：

@job(
    queue="default",       # 指定任务队列
    priority=1,            # 设置优先级
    max_attempts=3,        # 最大重试次数
    kwargs={"name": "World"}  # 默认参数
)
def greet(*, name: str):
    print(f"Hello, {name}!")

任务队列操作

提交任务到队列

创建任务后，需要将其推送到队列中才会被执行：

async with Chancy("postgresql://localhost/postgres") as chancy:
    await chancy.push(greet)

批量提交任务

Chancy支持一次性提交多个任务：

await chancy.push_many([job1, job2, job3])

任务状态追踪

提交任务后会返回一个引用对象，可用于追踪任务状态：

reference = await chancy.push(greet)
finished_job = await chancy.wait_for_job(reference)
print(finished_job.state)  # 查看任务最终状态

高级任务配置

任务优先级

通过设置优先级可以控制任务的执行顺序，数值越大优先级越高：

high_priority = greet.job.with_priority(10)   # 高优先级
low_priority = greet.job.with_priority(-10)   # 低优先级

任务重试机制

配置任务失败时的重试次数：

retry_job = greet.job.with_max_attempts(3)  # 最多重试3次

定时任务

可以安排任务在未来特定时间执行：

from datetime import datetime, timedelta, timezone

future_time = datetime.now(timezone.utc) + timedelta(hours=1)
scheduled_job = greet.job.with_scheduled_at(future_time)

注意：定时任务保证在指定时间或之后执行，但不保证精确时间。

资源限制

为任务设置执行时的资源限制：

from chancy import Limit, job

@job(limits=[
    Limit(Limit.Type.MEMORY, 1024 * 1024 * 1024),  # 1GB内存限制
    Limit(Limit.Type.TIME, 60),                    # 60秒时间限制
])
def resource_intensive_task():
    # 执行资源密集型操作
    pass

注意：不同执行器支持的限制类型可能不同。

唯一性任务

确保同一时间只有一个特定任务在执行：

unique_job = greet.job.with_unique_key("greet_alice")
await chancy.push(unique_job)

唯一性键保证同一时间只有一个具有相同键的任务处于排队或运行状态。

最佳实践

1. 合理设置优先级：关键任务应设置较高优先级，确保及时执行
2. 配置适当的重试次数：根据任务重要性设置合理的重试次数
3. 使用资源限制：防止单个任务消耗过多系统资源
4. 利用唯一性键：避免重复执行相同操作
5. 定时任务规划：合理安排非紧急任务的执行时间

通过掌握这些任务配置选项，您可以构建出高效、可靠的异步任务处理系统。