3. 第一个Celery任务

上一节我们已经理解了Celery的“生产者—Broker—消费者—结果后端”四段式结构。本节要把这套结构真正跑起来：写一个最小的任务模块，启动一个Worker进程，从另一个进程里调用任务，再通过结果后端取回执行结果。整个过程会暴露Celery最核心的运行机制——任务不是被“直接执行”的，而是被序列化、投递、领取、执行、回写，最后才被调用方读到。理解这条链路，比记住几个API更重要。

一、准备环境与目录结构

先约定一个最小可运行的工程目录。示例使用Redis同时充当Broker和Result Backend，这是本地开发最省事的组合；生产环境可以把两者拆开，但本节不展开。

celery_demo/
├── tasks.py          # 任务定义
├── producer.py       # 调用方（生产者）
└── celery_app.py     # Celery实例与配置

安装依赖：

pip install celery redis

确保本机Redis已启动，默认监听 127.0.0.1:6379，无密码。若Redis未启动，Worker会反复重连并报 Connection refused，这是新手最常见的第一个坑。

二、创建Celery实例

在 celery_app.py 中创建应用对象。Celery的第一个参数是当前模块名，用于生成默认的任务名前缀；broker 指定消息队列地址，backend 指定结果存储地址。

from celery import Celery

app = Celery(
    "celery_demo",
    broker="redis://127.0.0.1:6379/0",
    backend="redis://127.0.0.1:6379/1",
)

app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Shanghai",
    enable_utc=False,
)

这里把Broker放在db0、结果放在db1，是为了在Redis里能直观区分“待执行消息”和“执行结果”两类数据。序列化统一用JSON，避免pickle带来的跨语言与安全风险。

三、定义第一个任务

在 tasks.py 中导入app并注册任务。任务本质上就是一个被 @app.task 装饰的普通函数。

import time
from celery_app import app

@app.task(bind=True, max_retries=3)
def add(self, x, y):
    time.sleep(2)          # 模拟耗时操作
    return x + y

bind=True 让函数第一个参数变成任务实例 self，后续做重试、访问请求上下文时会用到。max_retries=3 表示最多重试3次。函数体里的 time.sleep(2) 是刻意加的，用来观察“调用立即返回、结果稍后才就绪”这一异步特征。

四、启动Worker

在 celery_demo 目录下执行：

celery -A celery_app worker --loglevel=info --concurrency=2

参数含义：-A celery_app 指定应用模块；--loglevel=info 输出任务接收与完成日志；--concurrency=2 启动2个执行槽位。启动成功后终端会打印类似：

[tasks]
  . celery_demo.tasks.add

[queues]
  . celery  exchange=celery(direct) key=celery

[2024-06-01 10:00:00,000: INFO/MainProcess] celery@host ready.

看到 ready. 说明Worker已连上Broker并开始监听队列。此时它处于阻塞等待状态，不会主动做任何事，直到有任务消息到达。

五、调用任务并查看结果

在另一个终端运行 producer.py：

from tasks import add

result = add.delay(3, 5)
print("task_id:", result.id)
print("state:", result.state)      # 立即查看，通常是 PENDING
print("result:", result.get(timeout=10))

add.delay(3, 5) 是 add.apply_async(args=(3, 5)) 的简写。它并不会在本进程执行加法，而是把 {"task": "celery_demo.tasks.add", "args": [3, 5], "id": "..."} 序列化后投递到Redis的celery队列，然后立刻返回一个 AsyncResult 对象。

典型输出：

task_id: 8f3c1e2a-...
state: PENDING
result: 8

注意 state 第一次打印是 PENDING，这不是“任务还没开始”，而是“结果后端里还没有这条任务的任何状态记录”。Celery对未知任务统一返回PENDING，这一点很容易被误解。result.get(timeout=10) 会阻塞轮询结果后端，直到拿到返回值或超时。

六、任务生命周期与执行链路

把上面的过程串起来，就是本节最需要建立的心智模型。下图展示了从调用到取回结果的完整链路，以及各阶段任务状态的变化。

生产者进程 add.delay(3, 5) 返回 AsyncResult Broker (Redis db0) celery 队列 消息持久化等待 Worker 进程 领取消息 → 执行 add concurrency=2 Result Backend Redis db1 存储状态与返回值 投递 领取 回写 result.get() 轮询读取结果 任务状态变化（以 add.delay 为例） PENDING 消息已投递 尚无状态记录 STARTED Worker 领取 开始执行 SUCCESS 返回 8 写入 Backend FAILURE 异常抛出 记录 traceback 投递后 执行中 执行结束 关键点：delay() 只负责投递，不等待执行；result.get() 才阻塞读取结果；PENDING 表示“后端无记录”，不等于“未开始”。 Worker 与生产者是两个独立进程，任务参数必须可 JSON 序列化，返回值同样经序列化后写入 Backend。

七、观察日志与结果判断

调用 add.delay(3, 5) 后，Worker终端会依次出现：

[INFO/MainProcess] Received task: celery_demo.tasks.add[8f3c1e2a-...]
[INFO/ForkPoolWorker-1] Task celery_demo.tasks.add[8f3c1e2a-...] succeeded in 2.01s: 8

这两行日志分别对应“领取”和“完成”两个节点。如果任务抛异常，第二行会变成 raised unexpected: ...，同时结果后端里该任务状态为FAILURE，result.get() 会重新抛出原异常。可以用 result.get(propagate=False) 拿到异常对象而不抛出，便于在调用方做统一错误处理。

判断任务是否真正跑通，不能只看 delay() 有没有报错，而要看三件事：Worker日志里是否出现 Received task 与 succeeded；result.get() 是否返回预期值；Redis db1 中是否存在以任务ID为键的结果记录。三者一致，才说明整条链路是通的。

调试阶段建议把 --loglevel 设为 debug，可以看到消息体内容与结果写入细节；稳定后调回 info，避免日志淹没关键信息。

八、常见失败与定位方向

第一个任务跑不起来，通常集中在四类问题：一是Broker地址写错或Redis未启动，Worker启动阶段就会报连接错误；二是任务模块未被Worker发现，日志里 [tasks] 列表为空，此时要检查 -A 指向的模块是否正确导入了任务；三是参数或返回值不可JSON序列化，比如传了datetime对象，会在投递或回写阶段报 EncodeError；四是调用方与Worker使用的Celery配置不一致，导致任务名对不上，消息被投到队列却无人领取。

定位时优先看Worker启动日志中的任务清单和队列名，再看调用后Worker是否打印 Received task。如果消息进了队列但Worker没反应，问题在路由或任务名；如果Worker收到了却执行失败，问题在任务函数本身或序列化。把这条判断顺序固定下来，后续排查复杂任务时会省很多时间。

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