本文共 1237 字，大约阅读时间需要 4 分钟。

简单使用

github：

新建tasks.py, 文件内容如下

# coding=utf-8from celery import Celery# celery 的相关配置celery = Celery('tasks', broker='redis://localhost:6379/0')# 具体任务(执行两个数相加)@celery.taskdef add(x, y):    return x + y

注册任务到任务队列

$ celery -A tasks worker -l info  ### -l info表示loglevel等级为info级，也可以设为debug级

• 输出如下：已经成功加入到task列表中了
  

调用task

# 打开python交互式环境$ python# 输入一下两行代码# tasks 为tasks.py 的路径(样例是在同一目录下)>>> from tasks import add>>> add.delay(2,5)# 返回结果的编号>>>
   

可以看到celery 控制台输出了任务执行的日志

实际应用（task与配置分离）

新建python 包 celeryTasks ,目录如下

celeryTasks/├── celery.py├── config.py├── __init__.py└── tasks.py

文件内容

celery.py

from __future__ import absolute_import from celery import Celery app = Celery('myCeleryTasks', include=['myCeleryTasks.tasks']) app.config_from_object('myCeleryTasks.config')

config.py

# config filefrom __future__ import absolute_importCELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'BROKER_URL = 'redis://127.0.0.1:6379/0'

tasks.py

from __future__ import absolute_importfrom myCeleryTasks.celery import app @app.taskdef add(x, y):     return x + y

注册task

$ celery -A myCeleryTasks worker -l info

调用

# 进入myCeleryTasks同级目录$ python #打开python交互式环境>>>from myCeleryTasks import add>>>add.delay(1,2)

输出效果和“简单使用”是一样的