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定時任務：