Django使用Celery異步任務隊列

1  Celery簡介

Celery是異步任務隊列，能夠獨立於主進程運行，在主進程退出後，也不影響隊列中的任務執行。

任務執行異常退出，從新啓動後，會繼續執行隊列中的其餘任務，同時能夠緩存中止期間接收的工做任務，這個功能依賴於消息隊列（MQ、Redis）。

 

1.1  Celery原理

 

 

Celery的架構由三部分組成，消息中間件（message broker），任務執行單元（worker）和任務執行結果存儲（task result store）組成。

• 消息中間件：Celery自己不提供消息服務，可是能夠方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis, MongoDB (experimental), Amazon SQS (experimental),CouchDB (experimental), SQLAlchemy (experimental),Django ORM (experimental), IronMQ。推薦使用：RabbitMQ、Redis做爲消息隊列。
• 任務執行單元：Worker是Celery提供的任務執行的單元，worker併發的運行在分佈式的系統節點中。
• 任務結果存儲：Task result store用來存儲Worker執行的任務的結果，Celery支持以不一樣方式存儲任務的結果，包括AMQP, Redis，memcached, MongoDB，SQLAlchemy, Django ORM，Apache Cassandra, IronCache

1.2     Celery適用場景

• 異步任務處理：例如給註冊用戶發送短消息或者確認郵件任務。
• 大型任務：執行時間較長的任務，例如視頻和圖片處理，添加水印和轉碼等，須要執行任務時間長。
• 定時執行的任務：支持任務的定時執行和設定時間執行。例如性能壓測定時執行。

 

2      Celery開發環境準備

2.1     環境準備

軟件名稱	版本號	說明
Linux	Centos 6.5（64bit）	操做系統
Python	3.5.2
Django	1.10	Web框架
Celery	4.0.2	異步任務隊列
Redis	2.4	消息隊列

 

2.2     Celery安裝

使用方法介紹：

Celery的運行依賴消息隊列，使用時須要安裝redis或者rabbit。

這裏咱們使用Redis。安裝redis庫:

sudo yum install redis

　　

啓動redis:

sudo service redis start

 

安裝celery庫

sudo pip install celery==4.0.2

 

3      Celery單獨執行任務

3.1     編寫任務

建立task.py文件

說明：這裏初始Celery實例時就加載了配置，使用的redis做爲消息隊列和存儲任務結果。

 

 

運行celery：

$ celery -A task worker --loglevel=info

看到下面的打印，說明celery成功運行。

 

3.2     調用任務

 直接打開python交互命令行

 執行下面代碼：

 

能夠celery的窗口看到任務的執行信息

 

 

任務執行狀態監控和獲取結果：

 

 

 

3.3     任務調用方法總結

有兩種方法：

delay和apply_async ，delay方法是apply_async簡化版。

add.delay(2, 2)
add.apply_async((2, 2))
add.apply_async((2, 2), queue='lopri')

 

delay方法是apply_async簡化版本。

apply_async方法是能夠帶很是多的配置參數，包括指定隊列等

• Queue 指定隊列名稱，能夠把不一樣任務分配到不一樣的隊列

 

3.4     任務狀態

每一個任務有三種狀態：

PENDING -> STARTED -> SUCCESS

 

任務查詢狀態：

res.state

 

來查詢任務的狀態

 

 

4      與Django集成

上面簡單介紹了celery異步任務的基本方法，結合咱們實際的應用，咱們須要與Django一塊兒使用，下面介紹如何與Django結合。

4.1     與Django集成方法

與Django集成有兩種方法：

• Django 1.8 以上版本：與Celery 4.0版本集成
• Django 1.8 如下版本：與Celery3.1版本集成，使用django-celery庫

 

今天咱們介紹celery4.0 和django 1.8以上版本集成方法。

4.2     建立項目文件

建立一個項目：名字叫作proj

- proj/
  - proj/__init__.py
  - proj/settings.py
  - proj/urls.py
  - proj/wsgi.py
- manage.py

 

建立一個新的文件：proj/proj/mycelery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
 
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
 
app = Celery('proj')
 
# 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()

 

在proj/proj/__init__.py:添加

from __future__ import absolute_import, unicode_literals
 
# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .mycelery import app as celery_app
 
__all__ = ['celery_app']

 

4.3     配置Celery

咱們在mycelery.py文件中說明celery的配置文件在settings.py中，而且是以CELERY開頭。

app.config_from_object('django.conf:settings', namespace='CELERY')

 

在settings.py文件中添加celery配置：

 

 

咱們的配置是使用redis做爲消息隊列，消息的代理和結果都是用redis，任務的序列化使用json格式。

重要：redis：//127.0.0.1:6379/0這個說明使用的redis的0號隊列，若是有多個celery任務都使用同一個隊列，則會形成任務混亂。最好是celery實例單獨使用一個隊列。

4.4     建立APP

建立Django的App，名稱爲celery_task，在app目錄下建立tasks.py文件。

完成後目錄結構爲：

├── celery_task
│   ├── admin.py
│   ├── apps.py
│   ├── __init__.py
│   ├── migrations
│   │   └── __init__.py
│   ├── models.py
│   ├── tasks.py
│   ├── tests.py
│   └── views.py
├── db.sqlite3
├── manage.py
├── proj
│   ├── celery.py
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── templates

 

4.5     編寫task任務

編輯任務文件

tasks.py

在tasks.py文件中添加下面代碼

# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def add(x, y):
    return x + y
 
@shared_task
def mul(x, y):
    return x * y
 
@shared_task
def xsum(numbers):
    return sum(numbers)

 

啓動celery:

celery -A proj.mycelery worker -l info

 

說明：proj爲模塊名稱，mycelery爲celery的實例所在的文件。

啓動成功打印：

 

 

4.6     在views中調用任務

在views中編寫接口，實現兩個功能：

• 觸發任務，而後返回任務的結果和任務ID
• 根據任務ID查詢任務狀態

代碼以下：

 

 

啓動django。

新開一個會話啓動celery；啓動命令爲：

celery –A proj.mycelery worker –l info

 

訪問http://127.0.0.1:8000/add，能夠看到返回的結果。

 

 

在celery運行的頁面，能夠看到下面輸出：

 

 

4.7     在views中查詢任務狀態

有的時候任務執行時間較長，須要查詢任務是否執行完成，能夠根據任務的id來查詢任務狀態，根據狀態進行下一步操做。

能夠看到任務的狀態爲：SUCCESS

 

 

5      Celery定時任務

Celery做爲異步任務隊列，咱們能夠按照咱們設置的時間，定時的執行一些任務，例如每日數據庫備份，日誌轉存等。

Celery的定時任務配置很是簡單：

定時任務的配置依然在setting.py文件中。

說明：若是以爲celery的數據配置文件和Django的都在setting.py一個文件中不方便，能夠分拆出來，只須要在mycelery.py的文件中指明便可。

app.config_from_object('django.conf:yoursettingsfile', namespace='CELERY')

 

 

5.1     任務間隔運行

#每30秒調用task.add
from datetime import timedelta

CELERY_BEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'tasks.add',
        'schedule': timedelta(seconds=30),
        'args': (16, 16)
    },
}

 

5.2     定時執行

定時天天早上7：30分運行。

注意：設置任務時間時注意時間格式，UTC時間或者本地時間。

#crontab任務
#天天7:30調用task.add
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    # Executes every Monday morning at 7:30 A.M
    'add-every-monday-morning': {
        'task': 'tasks.add',
        'schedule': crontab(hour=7, minute=30),
        'args': (16, 16),
    },
}

 

5.3     定時任務啓動

配置了定時任務，除了worker進程外，還須要啓動一個beat進程。

Beat進程的做用就至關於一個定時任務，根據配置來執行對應的任務。

5.3.1  啓動beat進程

命令以下：

celery -A proj.mycelery beat -l info

 

 

5.3.2  啓動worker進程

Worker進程啓動和前面啓動命令同樣。

 

celery –A proj.mycelery worker –l info

 

6      Celery深刻

Celery任務支持多樣的運行模式：

• 支持動態指定併發數 --autoscale=10,3 (always keep 3 processes, but grow to 10 if necessary).
• 支持鏈式任務
• 支持Group任務
• 支持任務不一樣優先級
• 支持指定任務隊列
• 支持使用eventlet模式運行worker

例如：指定併發數爲1000

celery -A proj.mycelery worker -c 1000

 

這些能夠根據使用的深刻自行了解和學習。

 

 

7      參考資料

Celery官網：

http://docs.celeryproject.org/en/latest/index.html

Celery與Django：

http://docs.celeryproject.org/en/latest/getting-started/next-steps.html#next-steps

celery定時任務：

http://blog.csdn.net/sicofield/article/details/50937338