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.2Celery適用場景

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

 2Celery開發環境準備

 2.1 環境準備

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

3Celery單獨執行任務

 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集成有兩種方法：

1. Django 1.8 以上版本：與Celery 4.0版本集成
2. 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_literalsimport osfrom 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 herefrom __future__ import absolute_import, unicode_literalsfrom celery import shared_task@shared_taskdef add(x, y): return x + y @shared_taskdef mul(x, y): return x * y @shared_taskdef xsum(numbers): return sum(numbers)

啟動celery:celery -A proj.mycelery worker -l info

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

啟動成功打?。?/p>

4.6在views中調用任務

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

1. 觸發任務，然后返回任務的結果和任務ID
2. 根據任務ID查詢任務狀態

代碼如下：

啟動django。

新開一個會話啟動celery；啟動命令為：celery –A proj.mycelery worker –l info

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

在celery運行的頁面，可以看到下面輸出：

4.7在views中查詢任務狀態

有的時候任務執行時間較長，需要查詢任務是否執行完成，可以根據任務的id來查詢任務狀態，根據狀態進行下一步操作。

可以看到任務的狀態為：SUCCESS

5Celery定時任務

Celery作為異步任務隊列，我們可以按照我們設置的時間，定時的執行一些任務，例如每日數據庫備份，日志轉存等。

Celery的定時任務配置非常簡單：

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

說明：如果覺得celery 的數據配置文件和Django 的都在setting.py 一個文件中不方便，可以分拆出來，只需要在mycelery.py 的文件中指明即可。

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

5.1任務間隔運行

#每30秒調用task.addfrom datetime import timedeltaCELERY_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.addfrom celery.schedules import crontabCELERY_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任務支持多樣的運行模式：

1. 支持動態指定并發數 --autoscale=10,3 (always keep 3 processes, but grow to 10 if necessary).
2. 支持鏈式任務
3. 支持Group任務
4. 支持任務不同優先級
5. 支持指定任務隊列
6. 支持使用eventlet模式運行worker

例如：指定并發數為1000

celery -A proj.mycelery worker -c 1000

這些可以根據使用的深入自行了解和學習。

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持VEVB武林網。