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