4. 任务定义与调用方式

在 Celery 中，一个 Python 函数并不会自动变成可被 Worker 执行的任务。真正让函数“升级”为任务的，是 @app.task 装饰器。它把普通函数包装成 Task 类的实例，并赋予它 delay()、apply_async()、s()、si() 等调用与组合能力。理解装饰器参数、两种调用方式的差异，以及签名与绑定机制，是写出可维护异步任务的基础。

一、@app.task 装饰器：把函数变成任务

最简形式如下：

from celery import Celery

app = Celery('demo', broker='redis://localhost:6379/0')

@app.task
def add(x, y):
    return x + y

此时 add 已不再是普通函数，而是 Task 实例。它既可以直接调用（同步执行，add(1, 2) 返回 3），也可以通过 add.delay(1, 2) 投递到 Broker 异步执行。装饰器常用参数包括：

  • name：任务注册名，默认是 模块名.函数名。显式指定可避免重构模块路径后旧消息找不到任务。
  • bind=True：把任务实例自身作为第一个参数传入，函数签名变为 (self, ...)，可访问 self.request、self.retry()。
  • base：自定义任务基类，用于统一日志、埋点、异常处理。
  • max_retries、default_retry_delay：默认重试次数与间隔。
  • acks_late：任务执行完才确认，配合幂等设计可提升可靠性。
  • rate_limit：如 '10/m'，限制该任务每分钟执行次数。
  • ignore_result：不保存返回值，减少 Backend 压力。
  • serializer：指定该任务参数序列化方式，如 'json'。

二、delay 与 apply_async 的区别

delay() 是 apply_async() 的简化封装，二者最终都走同一套消息投递逻辑，但能力范围不同：

对比项delay()apply_async()
参数传递仅位置参数与关键字参数支持 args、kwargs 显式分离
执行选项不支持支持 countdown、eta、expires、queue、priority、retry 等
适用场景立即投递、无特殊要求延迟、定时、指定队列、优先级控制
# 立即执行
add.delay(3, 5)

# 10 秒后执行，投递到 high 队列，1 小时后过期
add.apply_async(args=[3, 5], countdown=10,
                queue='high', expires=3600)

# 指定绝对时间执行
from datetime import datetime, timedelta
eta = datetime.utcnow() + timedelta(minutes=5)
add.apply_async(args=[3, 5], eta=eta)

需要延迟、定时、路由或优先级时，必须使用 apply_async；delay 只适合“立刻、默认队列、默认选项”的简单投递。

三、任务签名：把调用参数与执行选项解耦

签名（signature）是对“一次任务调用”的封装，用 s() 创建，可先构造、后投递，也可组合成工作流。常见变体：

  • add.s(3, 5)：普通签名，参数可被后续 partial 或链式调用覆盖。
  • add.si(3, 5)：不可变签名（immutable），参数锁定，链式调用时不会被上一任务结果替换。
  • add.signature(args=[3, 5], countdown=10)：等价写法，可携带执行选项。
from celery import chain

sig = add.s(3, 5)
sig.delay()                       # 等价于 add.delay(3, 5)

# 链式：先 add(3,5)=8，再把 8 作为 mul 的第一个参数
chain(add.s(3, 5), mul.s(10))()

# 不可变：mul 的参数固定为 (10, 2)，不受上一任务结果影响
chain(add.s(3, 5), mul.si(10, 2))()

签名让任务组合（chain、group、chord）成为可能，也是 Celery 工作流的核心构件。

四、不可变任务与绑定任务

不可变任务指签名使用 si() 或 immutable=True，其参数在链式调用中不会被前序任务结果替换。它适合“参数固定、不依赖上游返回值”的步骤，例如固定写日志、固定发通知。

绑定任务通过 bind=True 实现，任务函数第一个参数是 self，可访问任务运行时上下文：

@app.task(bind=True, max_retries=3, default_retry_delay=5)
def fetch(self, url):
    try:
        return requests.get(url, timeout=5).text
    except requests.RequestException as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

绑定任务的价值在于：self.request.id 获取任务 ID，self.request.retries 获取当前重试次数，self.retry() 触发重试并支持指数退避。它把“任务自身状态”暴露给业务逻辑，是编写可靠重试、幂等与可观测任务的关键。

选择建议：简单投递用 delay；需要延迟、路由、优先级用 apply_async；需要组合工作流用签名；需要重试与上下文用 bind=True；需要固定参数不被上游覆盖用 si()。
任务定义与调用方式全景 普通函数 def add(x, y): ... @app.task name / bind / base / retry Task 实例 delay / apply_async / s delay() 仅 args / kwargs 立即投递，默认队列 apply_async() countdown / eta / expires queue / priority / retry 签名 s() / si() 先构造，后投递 可组合 chain / group 普通签名 s() 参数可被上游结果覆盖 适合链式传递 不可变签名 si() 参数锁定，不被覆盖 适合固定步骤 绑定任务 bind=True self.request / self.retry 重试与上下文 选择路径：简单投递 → delay；延迟/路由/优先级 → apply_async；工作流组合 → 签名；重试与上下文 → bind=True；固定参数 → si() 所有调用最终都经 Broker 投递，由 Worker 消费执行

把上述机制串起来看：@app.task 决定任务“是什么”，delay 与 apply_async 决定“怎么投”，签名决定“怎么组合”，bind 与 si 决定“参数与上下文如何被处理”。在实际工程中，建议为每个任务显式命名、按需开启 bind 与重试、对固定步骤使用 si()，并在需要延迟或路由时统一使用 apply_async，从而让任务定义清晰、调用可控、工作流可组合。

← 上一章 📖 返回目录 下一章 →