celery+django+redis使用介绍

版本：

• celery==3.1.25
• django==1.8.6
• redis==2.10.6

安装：

• 进入虚拟环境（虚拟环境创建不同，进入方式不同）
• pip install celery==3.1.25（以celery安装为例，其他安装方式相同）

运行环境：

• window10（celery4.0以后不支持windows）
• linux

目录结构：

• 最外面的test1为项目名称
• 里面的test1与app同级目录，里面都是一些配置（其中celery.py放在其中）。
• app2为项目中的一个app
• tasks.py在app2中，这个tasks.py只针对app2使用。若其他app也有celery任务。建立同样的文件名即可。

运行celery，需要的几个文件：

• test1下的celery.py，代码如下（里面有介绍）：

from __future__ import absolute_import, unicode_literals
import os
from django.conf import settings
from celery import Celery


#设置 Django 的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'test.settings')

# 创建 celery 实例
app = Celery('test1')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')

# 搜索所有 app 中的 tasks
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
 
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))

• test1下的settings.py，代码如下：

"""原来的django的settings内容"""

#celery config
#消息中间件（使用redis），消息代理，用于发布者传递消息给消费者
BROKER_URL = 'redis://127.0.0.1:6379'
#消息结果返回中间件（使用redis），用于存储任务执行结果
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379'
#允许的内容类型，
CELERY_ACCEPT_CONTENT = ['json']
#任务的序列化方式
CELERY_TASK_SERIALIZER = 'json'
#任务结果的序列化方式
CELERY_RESULT_SERIALIZER = 'json'
#celery时区，定时任务使用
CELERY_TIMEZONE = 'Asia/Shanghai'
from datetime import timedelta
#定时任务处理，使用的schedule，里面的task，写上任务的“路径”，schedule设置时间，args设置参数。
CELERYBEAT_SCHEDULE = {
'add_every_10_seconds': {
'task': 'app2.tasks.add',
'schedule': timedelta(seconds=10),
'args': (4,4)
},
}

• test1下的__init__.py，代码如下：

from __future__ import absolute_import
from .celery import app as celery_app
# 这是为了确保在django启动时启动 celery

• app2下的tasks.py,这个tasks文件只针对该app2使用，其他app可以根据实际情况添加tasks.py文件，里面写相应的任务代码。代码如下：

# -*- coding:utf-8 -*-
from __future__ import absolute_import
from auto_model_platform.celery import app

@app.task
def add(x, y):
    time.sleep(30)
    print("running...", x, y)
    return x + y

调用tasks.py内的函数:

• 在app2.views.py中的某个视图函数中直接调用，比如调用tasks.py中的add（x,y）函数，这样调用。代码如下：

from app2.tasks import add

class TestView(View):
     def get(self, request):

    """其它逻辑"""

    #celery处理的其它任务（异步处理）,下面这个代码，celery会去处理，django直接执行下面的其它逻辑
    r = add.delay(x,y)
    task_id = r.id

     """其它逻辑"""

    return JsonResponse({"data":"123"})

• 通过task_id可以查询celery异步处理完的结果

from test1.celery import app
status = app.AsyncResult(task_id).status
result = app.AsyncResult(task_id).result

#状态有这几种情况
CELERY_STATUS = {
    'PENDING': '等待开始',
    'STARTED': '任务开始',
    'SUCCESS': '成功',
    'FAILURE': '失败',
    'RETRY': '重试',
    'REVOKED': '任务取消',
}

服务器启动celery worker（消费者）任务，和定时任务:

• 打开项目的虚拟环境，进入项目名称目录下，即跟app同一级目录下运行：
• 定时任务：celery -A auto_model_platform worker -l info --beat
• worker任务：celery -A auto_model_platform worker -l info