原生celery，非djcelery模組，所有演示均基於Django2.0

celery是一個基於python開發的簡單、靈活且可靠的分散式任務佇列框架，支援使用任務佇列的方式在分散式的機器/程序/執行緒上執行任務排程。採用典型的生產者-消費者模型，主要由三部分組成：
1. 訊息佇列broker：broker實際上就是一個MQ佇列服務，可以使用redis、rabbitmq等作為broker
2. 處理任務的消費者workers：broker通知worker佇列中有任務，worker去佇列中取出任務執行，每一個worker就是一個程序
3. 儲存結果的backend：執行結果儲存在backend，預設也會儲存在broker使用的MQ佇列服務中，也可以單獨配置用何種服務做backend

非同步任務

我的非同步使用場景為專案上線：前端web上有個上線按鈕，點選按鈕後發請求給後端，後端執行上線過程要5分鐘，後端在接收到請求後把任務放入佇列非同步執行，同時馬上返回給前端一個任務執行中的結果。若果沒有非同步執行會怎麼樣呢？同步的情況就是執行過程中前端一直在等後端返回結果，頁面轉呀轉的就轉超時了。

非同步任務配置

1.安裝rabbitmq，這裡我們使用rabbitmq作為broker，安裝完成後預設啟動了，也不需要其他任何配置

# apt-get install rabbitmq-server

2.安裝celery

# pip3 install celery
 

3.celery用在django專案中，django專案目錄結構(簡化)如下

website/
|-- deploy
|   |-- admin.py
|   |-- apps.py
|   |-- __init__.py
|   |-- models.py
|   |-- tasks.py
|   |-- tests.py
|   |-- urls.py
|   `-- views.py
|-- manage.py
|-- README
`-- website
    |-- celery.py
    |-- __init__.py
    |-- settings.py
    |
 
-- urls.py
    `-- wsgi.py

4.建立website/celery.py主檔案

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery, platforms

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'website.settings')

app = Celery('website')

# Using a string here means the worker don't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

# 允許root 使用者執行celery
platforms.C_FORCE_ROOT = True

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

5.在website/__init__.py檔案中增加如下內容，確保django啟動的時候這個app能夠被載入到

from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ['celery_app']

6.各應用建立tasks.py檔案，這裡為deploy/tasks.py

from __future__ import absolute_import
from celery import shared_task

@shared_task
def add(x, y):
    return x + y

• 注意tasks.py必須建在各app的根目錄下，且只能叫tasks.py，不能隨意命名

7.views.py中引用使用這個tasks非同步處理

from deploy.tasks import add

def post(request):
    result = add.delay(2, 3)

• 使用函式名.delay()即可使函式非同步執行
• 可以通過result.ready()來判斷任務是否完成處理
• 如果任務丟擲一個異常，使用result.get(timeout=1)可以重新丟擲異常
• 如果任務丟擲一個異常，使用result.traceback可以獲取原始的回溯資訊

8.啟動celery

# celery -A website worker -l info

9.這樣在呼叫post這個方法時，裡邊的add就可以非同步處理了

定時任務

定時任務的使用場景就很普遍了，比如我需要定時傳送報告給老闆~

定時任務配置

1.website/celery.py檔案新增如下配置以支援定時任務crontab

from celery.schedules import crontab

app.conf.update(
    CELERYBEAT_SCHEDULE = {
        'sum-task': {
            'task': 'deploy.tasks.add',
            'schedule':  timedelta(seconds=20),
            'args': (5, 6)
        }
        'send-report': {
            'task': 'deploy.tasks.report',
            'schedule': crontab(hour=4, minute=30, day_of_week=1),
        }
    }
)

• 定義了兩個task：

  • 名字為’sum-task’的task，每20秒執行一次add函式，並傳了兩個引數5和6
  • 名字為’send-report’的task，每週一早上4：30執行report函式

• timedelta是datetime中的一個物件，需要from datetime import timedelta引入，有如下幾個引數

  • days：天
  • seconds：秒
  • microseconds：微妙
  • milliseconds：毫秒
  • minutes：分
  • hours：小時

• crontab的引數有：

  • month_of_year：月份
  • day_of_month：日期
  • day_of_week：周
  • hour：小時
  • minute：分鐘

2.deploy/tasks.py檔案新增report方法：

@shared_task
def report():
    return 5

3.啟動celery beat，celery啟動了一個beat程序一直在不斷的判斷是否有任務需要執行

# celery -A website beat -l info

Tips

1.如果你同時使用了非同步任務和計劃任務，有一種更簡單的啟動方式celery -A website worker -b -l info，可同時啟動worker和beat
2.如果使用的不是rabbitmq做佇列那麼需要在主配置檔案中website/celery.py配置broker和backend，如下：

# redis做MQ配置
app = Celery('website', backend='redis', broker='redis://localhost')
# rabbitmq做MQ配置
app = Celery('website', backend='amqp', broker='amqp://admin:[email protected]')

3.celery不能用root使用者啟動的話需要在主配置檔案中新增platforms.C_FORCE_ROOT = True
4.celery在長時間執行後可能出現記憶體洩漏，需要新增配置CELERYD_MAX_TASKS_PER_CHILD = 10，表示每個worker執行了多少個任務就死掉

參考文章：