Django配置celery執行異步任務和定時任務

原生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:admin@localhost')
複製代碼

3.celery不能用root用戶啓動的話須要在主配置文件中添加platforms.C_FORCE_ROOT = True 4.celery在長時間運行後可能出現內存泄漏，須要添加配置CELERYD_MAX_TASKS_PER_CHILD = 10，表示每一個worker執行了多少個任務就死掉

參考文章：

• docs.celeryproject.org/en/latest/
• github.com/pylixm/cele…
• pylixm.cc/posts/2015-…