一文搞懂Celery任务队列与Worker原理（Django环境）

一、 简介

Celery 是一个简单、灵活且足够可靠的分布式消息处理系统，核心是一个专注于实时处理的任务队列，同时也支持任务调度。说白了，它就是为了帮你把那些耗时的操作从主流程里抽出来，异步执行，或者定时执行。

安装也很简单，一行命令搞定：

pip install celery

# 看看是否安装成功
pip freeze | grep -i 'celery'

二、名词解释

在真正上手之前，先理清几个关键角色：

1. broker——消息传输的中间件。生产者一旦有消息发送，就会先放到 broker 里。常见的实现有 Redis、RabbitMQ。
2. backend——用来存储任务执行的结果。如果你需要跟踪任务状态、查询返回值，就得配置 backend，比如 Redis 或 MySQL。
3. worker——真正干活的人。它是一个进程，负责从 broker 里取出消息并执行对应的任务。

三、使用 Celery

写一个简单的 Celery 应用只需要几行代码：

from celery import Celery

app = Celery('kqlproject', broker='redis://:password@127.0.0.1:6379/1')
# 第一个参数是自定义的项目名称

# 定义任务函数
@app.task
def task_test():
    print("task is running")

注意 broker 的地址要替换成你自己的 Redis 连接信息，密码可选。

四、启动 Worker

在终端里，进入 tasks.py 所在的目录，执行下面的命令来启动 worker：

# 前台启动，日志会直接输出到终端
celery -A tasks worker --loglevel=info

这时 worker 就处于监听状态，等着处理消息。

五、创建生产者——推送任务

在同一个目录下打开 Python 交互式环境（比如 ipython3），然后：

from tasks import task_test
task_test.delay()

执行之后，回到 worker 的终端窗口，你就能看到 task is running 的日志输出。

六、存储执行结果

Celery 也支持把任务执行结果保存下来，方便后续查询。需要借助 Redis、MySQL 或 Memcached 等后端存储。配置方法很简单，在创建 Celery 实例时加上 backend 参数：

from celery import Celery

app = Celery(
   'demo',
   broker='redis://@127.0.0.1:6379/1',
   backend='redis://@127.0.0.1:6379/2',
)

@app.task
def test_task(a, b):
    print("task is running")
    return a + b

这样任务执行完毕后，返回值就会存入指定的 Redis 数据库中。

七、Django 中使用 Celery

在 Django 项目里集成 Celery 也很规范，大致分四步走：

1. 在项目同名目录下创建 celery.py 配置文件。
2. 在各个应用下创建 tasks.py，集中定义要执行的 worker 函数。
3. 视图函数充当生产者，调用 delay() 推送任务。
4. 在项目根目录启动 worker。
启动命令示例：

celery -A 项目同名目录名 worker -l info

下面给出一个典型的 celery.py 配置：

# 在 settings 同级目录下新建 celery.py

from celery import Celery
from django.conf import settings
import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'dadablog.settings')

app = Celery('dadablog')
app.conf.update(
    BROKER_URL='redis://:@127.0.0.1:6379/1'
)
# 自动去注册的应用下寻找加载 worker 函数
app.autodiscover_tasks(settings.INSTALLED_APPS)

然后在每个应用下创建 tasks.py，比如发送信息的示例：

# 在某应用下的 tasks.py

from tools.sms import YunTongXin
from dadablog.celery import app

@app.task
def send_sms_c(phone, code):
    config = {
        "accountSid": "2c94811c***********",
        "accountToken": "bd81141c************8",
        "appId": "2c94811c88b*****************",
        "templateId": "1"
    }
    yun = YunTongXin(**config)
    res = yun.run(phone, code)
    return res

启动 worker 后，当视图函数调用 send_sms_c.delay(phone, code) 时，worker 就会异步执行信息发送任务。

八、正式环境后台启动

生产环境不能让 worker 挂在前台，需要后台运行。推荐使用 nohup 配合协程模式：

nohup celery -A projectname worker -P gevent -c 1000 > celery.log 2>&1 &

# 参数说明：
# nohub          —— 忽略所有挂断信号（SIGHUP）
# projectname    —— 项目配置目录名
# -P gevent      —— 使用 Python 协程，这里开启 1000 个协程
# celery.log     —— 日志输出文件（也可用绝对路径）
# 2>&1          —— 将标准错误重定向到标准输出
# &              —— 后台运行

这样即便终端关闭，worker 也能继续跑。

九、总结

Celery 的核心价值在于把耗时的任务从业务主流程中解耦出来，让系统响应更快、更稳定。从简单的单机任务到复杂的分布式调度，这套机制都能很好地胜任。希望这篇笔记能帮你快速上手，在实际项目中用好 Celery。