Celery異步處理

1.Celery概述

1.1問題拋出

我們在做網站后端程序開發時，會碰到這樣的需求：用戶需要在我們的網站填寫注冊信息，我們發給用戶一封注冊激活郵件到用戶郵箱，如果由於各種原因，這封郵件發送所需時間較長，那么客戶端將會等待很久，造成不好的用戶體驗.

那么怎么解決這樣的問題呢?

 

我們將耗時任務放到后台異步執行。不會影響用戶其他操作。除了注冊功能，例如上傳，圖形處理等等耗時的任務，都可以按照這種思路來解決。 如何實現異步執行任務呢？

我們可使用celery. celery除了剛才所涉及到的異步執行任務之外，還可以實現定時處理某些任務。

1.2celery介紹

Celery是一個功能完備即插即用的任務隊列。它使得我們不需要考慮復雜的問題，使用非常簡單。celery看起來似乎很龐大，本章節我們先對其進行簡單的了解，然后再去學習其他一些高級特性。 
celery適用異步處理問題，當發送郵件、或者文件上傳, 圖像處理等等一些比較耗時的操作，我們可將其異步執行，這樣用戶不需要等待很久，提高用戶體驗。 celery的特點是：

簡單，易於使用和維護，有豐富的文檔。
高效，單個celery進程每分鍾可以處理數百萬個任務。
靈活，celery中幾乎每個部分都可以自定義擴展。
celery非常易於集成到一些web開發框架中.

 2、快速入門（一）

 2.1任務隊列

 

  任務隊列是一種跨線程、跨機器工作的一種機制.

  任務隊列中包含稱作任務的工作單元。有專門的工作進程持續不斷的監視任務隊列，並從中獲得新的任務並處理.

  celery通過消息進行通信，通常使用一個叫Broker(中間人)來協client(任務的發出者)和worker(任務的處理者). clients發出消息到隊列中，broker將隊列中的信息派發給worker來處理。

  一個celery系統可以包含很多的worker和broker，可增強橫向擴展性和高可用性能。

2.2Celery安裝

我們可以使用python的包管理器pip來安裝:

pip install -U Celery
也可從官方直接下載安裝包:https://pypi.python.org/pypi/celery/

tar xvfz celery-0.0.0.tar.gz
cd celery-0.0.0
python setup.py build
python setup.py install

 

 

 2.3Broker(中間人)

Celery需要一種解決消息的發送和接受的方式，我們把這種用來存儲消息的的中間裝置叫做message broker, 也可叫做消息中間人。 作為中間人，我們有幾種方案可選擇：

RabbitMQ

RabbitMQ是一個功能完備，穩定的並且易於安裝的broker. 它是生產環境中最優的選擇。使用RabbitMQ的細節參照以下鏈接： http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmq

如果我們使用的是Ubuntu或者Debian發行版的Linux，可以直接通過下面的命令安裝RabbitMQ: sudo apt-get install rabbitmq-server 安裝完畢之后，RabbitMQ-server服務器就已經在后台運行。如果您用的並不是Ubuntu或Debian, 可以在以下網址： http://www.rabbitmq.com/download.html 去查找自己所需要的版本軟件。

Redis

Redis也是一款功能完備的broker可選項，但是其更可能因意外中斷或者電源故障導致數據丟失的情況。 關於是有那個Redis作為Broker，可訪下面網址： http://docs.celeryproject.org/en/latest/getting-started/brokers/redis.html#broker-redis

 

2.4應用

 使用celery第一件要做的最為重要的事情是需要先創建一個Celery實例，我們一般叫做celery應用，或者更簡單直接叫做一個app。app應用是我們使用celery所有功能的入口，比如創建任務，管理任務等，在使用celery的時候，app必須能夠被其他的模塊導入。

創建應用

我們首先創建tasks.py模塊, 其內容為:

from celery import Celery # 我們這里案例使用redis作為broker app = Celery('demo', broker='redis://:332572@127.0.0.1/1') # 創建任務函數 @app.task def my_task(): print("任務函數正在執行....") 

  Celery第一個參數是給其設定一個名字， 第二參數我們設定一個中間人broker, 在這里我們使用Redis作為中間人。my_task函數是我們編寫的一個任務函數， 通過加上裝飾器app.task, 將其注冊到broker的隊列中。

  現在我們在創建一個worker， 等待處理隊列中的任務.打開終端，cd到tasks.py同級目錄中，執行命令:

celery -A tasks worker --loglevel=info

顯示效果如下: 

 

 

調用任務

  任務加入到broker隊列中，以便剛才我們創建的celery workder服務器能夠從隊列中取出任務並執行。如何將任務函數加入到隊列中，可使用delay()。

進入python終端, 執行如下代碼:

from tasks import my_task
my_task.delay()

執行效果如下:

 

 我們通過worker的控制台，可以看到我們的任務被worker處理。調用一個任務函數，將會返回一個AsyncResult對象，這個對象可以用來檢查任務的狀態或者獲得任務的返回值。

存儲結果

  如果我們想跟蹤任務的狀態，Celery需要將結果保存到某個地方。有幾種保存的方案可選:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。

  例子我們仍然使用Redis作為存儲結果的方案，任務結果存儲配置我們通過Celery的backend參數來設定。我們將tasks模塊修改如下:

from celery import Celery # 我們這里案例使用redis作為broker app = Celery('demo', backend='redis://:332572@127.0.0.1:6379/2', broker='redis://:332572@127.0.0.1:6379/1') # 創建任務函數 @app.task def my_task(a, b): print("任務函數正在執行....") return a + b 

  我們給Celery增加了backend參數，指定redis作為結果存儲,並將任務函數修改為兩個參數，並且有返回值。 

 

更多關於result對象信息，請參閱下列網址:http://docs.celeryproject.org/en/latest/reference/celery.result.html#module-celery.result

 

 2.5配置

 

Celery使用簡單，配置也非常簡單。Celery有很多配置選項能夠使得celery能夠符合我們的需要，但是默認的幾項配置已經足夠應付大多數應用場景了。

  配置信息可以直接在app中設置，或者通過專有的配置模塊來配置。

直接通過app來配置

from celery import Celery app = Celery('demo') # 增加配置 app.conf.update( result_backend='redis://:332572@127.0.0.1:6379/2', broker_url='redis://:332572@127.0.0.1:6379/1', ) 

專有配置文件

  對於比較大的項目，我們建議配置信息作為一個單獨的模塊。我們可以通過調用app的函數來告訴Celery使用我們的配置模塊。

  配置模塊的名字我們取名為celeryconfig, 這個名字不是固定的，我們可以任意取名，建議這么做。我們必須保證配置模塊能夠被導入。 配置模塊的名字我們取名為celeryconfig, 這個名字不是固定的，我們可以任意取名，建議這么做。我們必須保證配置模塊能夠被導入。

  下面我們在tasks.py模塊 同級目錄下創建配置模塊celeryconfig.py:

result_backend = 'redis://:332572@127.0.0.1:6379/2' broker_url = 'redis://:332572@127.0.0.1:6379/1' 

  tasks.py文件修改為:

from celery import Celery import celeryconfig # 我們這里案例使用redis作為broker app = Celery('demo') # 從單獨的配置模塊中加載配置 app.config_from_object('celeryconfig') 

更多配置: http://docs.celeryproject.org/en/latest/userguide/configuration.html#configuration

 3、快速入門（二）

 3.1項目中使用celery

  我的項目目錄:

TestCelery/ ├── proj │ ├── celeryconfig.py │ ├── celery.py │ ├── init.py │ └── tasks.py └── test.py

  celery.py內容如下:

from celery import Celery

# 創建celery實例
app = Celery('demo')
app.config_from_object('proj.celeryconfig')

# 自動搜索任務
app.autodiscover_tasks(['proj'])
  celeryconfig.p模塊內容如下：

from kombu import Exchange, Queue
BROKER_URL = 'redis://:332572@127.0.0.1:6379/1'
CELERY_RESULT_BACKEND = 'redis://:332572@127.0.0.1:6379/2'
  tasks.py模塊內容如下:

from proj.celery import app as celery_app

# 創建任務函數
@celery_app.task
def my_task1():
    print("任務函數(my_task1)正在執行....")

@celery_app.task
def my_task2():
    print("任務函數(my_task2)正在執行....")

@celery_app.task
def my_task3():
    print("任務函數(my_task3)正在執行....")
  啟動worker:

celery -A proj worker -l info

 

鍵入ctrl+c可關閉worker.

 

 3.2調度任務

調用任務，可使用delay()方法:

my_task.delay(2, 2)
  也可以使用apply_async()方法，該方法可讓我們設置一些任務執行的參數，例如，任務多久之后才執行，任務被發送到那個隊列中等等.

my_task.apply_async((2, 2), queue='my_queue', countdown=10)
任務my_task將會被發送到my_queue隊列中，並且在發送10秒之后執行。

  如果我們直接執行任務函數，將會直接執行此函數在當前進程中，並不會向broker發送任何消息。

  無論是delay()還是apply_async()方式都會返回AsyncResult對象，方便跟蹤任務執行狀態，但需要我們配置result_backend.

  每一個被吊用的任務都會被分配一個ID，我們叫Task ID.

 

 

 3.3Desining Work-flows

 

signature

  我們到目前為止只是知道如何使用delay()方法，當然這個方法也是非常常用的。但是有時我們並不想簡單的將任務發送到隊列中，我們想將一個任務函數(由參數和執行選項組成)作為一個參數傳遞給另外一個函數中，為了實現此目標，Celery使用一種叫做signatures的東西。

  一個signature包裝了一個參數和執行選項的單個任務調用。我們可將這個signature傳遞給函數。

  我們先看下tasks.py模塊中定義的任務函數:

from proj.celery import app as celery_app # 創建任務函數 @celery_app.task def my_task1(): print("任務函數(my_task1)正在執行....") @celery_app.task def my_task2(): print("任務函數(my_task2)正在執行....") @celery_app.task def my_task3(): print("任務函數(my_task3)正在執行....") 

  我們將my_task1()任務包裝稱一個signature:

t1 = my_task1.signatures(countdown=10) t1.delay() 

Primitives

  這些primitives本身就是signature對象，因此它們可以以多種方式組合成復雜的工作流程。primitives如下:

  group: 一組任務並行執行，返回一組返回值，並可以按順序檢索返回值。

  chain: 任務一個一個執行，一個執行完將執行return結果傳遞給下一個任務函數.

  tasks.py模塊如下:

from proj.celery import app as celery_app # 創建任務函數 @celery_app.task def my_task1(a, b): print("任務函數(my_task1)正在執行....") return a + b @celery_app.task def my_task2(a, b): print("任務函數(my_task2)正在執行....") return a + b @celery_app.task def my_task3(a, b): print("任務函數(my_task3)正在執行....") return a + b 

  group案例如下(test.py模塊):

from proj.tasks import my_task1 from proj.tasks import my_task2 from proj.tasks import my_task3 from celery import group # 將多個signature放入同一組中 my_group = group((my_task1.s(10, 10), my_task2.s(20, 20), my_task3.s(30, 30))) ret = my_group() # 執行組任務 print(ret.get()) # 輸出每個任務結果

 

chain案例如下(test.py模塊):

from proj.tasks import my_task1 from proj.tasks import my_task2 from proj.tasks import my_task3 from celery import chain # 將多個signature組成一個任務鏈 # my_task1的運行結果將會傳遞給my_task2 # my_task2的運行結果會傳遞給my_task3 my_chain = chain(my_task1.s(10, 10) | my_task2.s(20) | my_task3.s(30)) ret = my_chain() # 執行任務鏈 print(ret.get()) # 輸出最終結果 

3.4Routing

 

 假如我們有兩個worker,一個worker專門用來處理郵件發送任務和圖像處理任務，一個worker專門用來處理文件上傳任務。

  我們創建兩個隊列，一個專門用於存儲郵件任務隊列和圖像處理，一個用來存儲文件上傳任務隊列。

  Celery支持AMQP(Advanced Message Queue)所有的路由功能，我們也可以使用簡單的路由設置將指定的任務發送到指定的隊列中.

  我們需要配置在celeryconfig.py模塊中配置 CELERY_ROUTES 項, tasks.py模塊修改如下:

from proj.celery import app as celery_app


@celery_app.task
def my_task1(a, b):
    print("my_task1任務正在執行....")
    return a + b


@celery_app.task
def my_task2(a, b):
    print("my_task2任務正在執行....")
    return a + b


@celery_app.task
def my_task3(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task4(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task5():
    print("my_task5任務正在執行....")


@celery_app.task
def my_task6():
    print("my_task6任務正在執行....")


@celery_app.task
def my_task7():
    print("my_task7任務正在執行....")
  我們通過配置，將send_email和upload_file任務發送到queue1隊列中，將image_process發送到queue2隊列中。

  我們修改celeryconfig.py:

broker_url='redis://:@127.0.0.1:6379/1'
result_backend='redis://:@127.0.0.1:6379/2'


task_routes=({
    'proj.tasks.my_task5': {'queue': 'queue1'},
    'proj.tasks.my_task6': {'queue': 'queue1'},
    'proj.tasks.my_task7': {'queue': 'queue2'},
    },
)
  test.py:

from proj.tasks import *

# 發送任務到路由指定的隊列中
my_task5.delay() my_task6.delay() my_task7.delay()

  開啟兩個worker服務器，分別處理兩個隊列：
```python
celery -A proj worker --loglevel=info -Q queue1
celery -A proj worker --loglevel=info -Q queue2
  我們同樣也可以通過apply_aynsc()方法來設置任務發送到那個隊列中:

my_task1.apply_async(queue='queue1')
  我們也可設置一個worker服務器處理兩個隊列中的任務:

celery -A proj worker --loglevel=info -Q queue1,queue2

4.Periodic Tasks(周期性任務)

celery beat是一個調度器，它可以周期內指定某個worker來執行某個任務。如果我們想周期執行某個任務需要增加beat_schedule配置信息.  

broker_url='redis://:@127.0.0.1:6379/1'
result_backend='redis://:@127.0.0.1:6379/2'

# 指定任務發到那個隊列中
task_routes=({
    'proj.tasks.my_task5': {'queue': 'queue1'},
    'proj.tasks.my_task6': {'queue': 'queue1'},
    'proj.tasks.my_task7': {'queue': 'queue2'},
    },
)


# 配置周期性任務，　或者定時任務
beat_schedule = {
    'every-5-seconds':
        {
            'task': 'proj.tasks.my_task8',
            'schedule': 5.0,
            # 'args': (16, 16),
        }
}
  tasks.py模塊內容如下:

from proj.celery import app as celery_app


@celery_app.task
def my_task1(a, b):
    print("my_task1任務正在執行....")
    return a + b


@celery_app.task
def my_task2(a, b):
    print("my_task2任務正在執行....")
    return a + b


@celery_app.task
def my_task3(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task4(a, b):
    print("my_task3任務正在執行....")
    return a + b


@celery_app.task
def my_task5():
    print("my_task5任務正在執行....")




@celery_app.task
def my_task6():
    print("my_task6任務正在執行....")



@celery_app.task
def my_task7():
    print("my_task7任務正在執行....")


# 周期執行任務
@celery_app.task
def my_task8():
    print("my_task8任務正在執行....")
  啟動woker處理周期性任務:

celery -A proj worker --loglevel=info --beat
  如果我們想指定在某天某時某分某秒執行某個任務，可以執行cron任務, 增加配置信息如下:

beat_schedule = {
    'every-5-minute':
        {
            'task': 'proj.tasks.period_task',
            'schedule': 5.0,
            'args': (16, 16),
        },
    'add-every-monday-morning': {
        'task': 'proj.tasks.period_task',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (16, 16),
    },

}
crontab例子: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

  開啟一個celery beat服務：

celery -A proj beat
  celery需要保存上次任務運行的時間在數據文件中，文件在當前目錄下名字叫celerybeat-schedule. beat需要訪問此文件：

celery -A proj beat -s /home/celery/var/run/celerybeat-schedule

 

 

 5.Django使用Celery

5.1配置celery

1. 配置celery
  創建django項目celery_demo, 創建應用demo:

django-admin startproject celery_demo
python manage.py startapp demo

 

  在celery_demo模塊中創建celery.py模塊, 文件目錄為:

　　

 

   celery.py模塊內容為:

from celery import Celery
from django.conf import settings
import os

# 為celery設置環境變量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings')

# 創建應用
app = Celery("demo")
# 配置應用
app.conf.update(
    # 配置broker, 這里我們用redis作為broker
    BROKER_URL='redis://:332572@127.0.0.1:6379/1',
)
# 設置app自動加載任務
# 從已經安裝的app中查找任務
app.autodiscover_tasks(settings.INSTALLED_APPS)
  在應用demo引用創建tasks.py模塊, 文件目錄為: 

 

  我們在文件內創建一個任務函數my_task:

from celery_demo.celery import app
import time

# 加上app對象的task裝飾器
# 此函數為任務函數
@app.task
def my_task():
    print("任務開始執行....")
    time.sleep(5)
    print("任務執行結束....")
  在views.py模塊中創建視圖index:

from django.shortcuts import render
from django.http import HttpResponse
from .tasks import my_task


def index(request):
# 將my_task任務加入到celery隊列中
# 如果my_task函數有參數，可通過delay()傳遞
# 例如 my_task(a, b), my_task.delay(10, 20)
    my_task.delay()

    return HttpResponse("<h1>服務器返回響應內容!</h1>")
  在celey_demo/settings.py配置視圖路由:

from django.conf.urls import url
from django.contrib import admin
from demo.views import index

urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^$', index),
]
  創建worker等待處理celery隊列中任務, 在終端執行命令:

celery -A celery_demo worker -l info

 

  啟動django測試服務器：

python manage.py runserver

 

 

 5.2保存任務結果

此處需要用到額外包django_celery_results, 先安裝包:

pip install django-celery-results
  在celery_demo/settings.py中安裝此應用:

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'demo',
    'django_celery_results',  # 注意此處應用名為下划線
]
  回到celery_demo/celery.py模塊中，增加配置信息如下:

from celery import Celery
from django.conf import settings
import os

# 為celery設置環境變量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings')

# 創建應用
app = Celery("demo")
# 配置應用
app.conf.update(
    # 配置broker, 這里我們用redis作為broker
    BROKER_URL='redis://:332572@127.0.0.1:6379/1',
    # 使用項目數據庫存儲任務執行結果
    CELERY_RESULT_BACKEND='django-db',
)
# 設置app自動加載任務
# 從已經安裝的app中查找任務
app.autodiscover_tasks(settings.INSTALLED_APPS)
  創建django_celery_results應用所需數據庫表, 執行遷移文件：

python manage.py migrate django_celery_results
  我這里使用的是django默認的數據庫sqlit, 執行遷移之后，會在數據庫中創建一張用來存儲任務結果的表: 

  再次從瀏覽器發送請求， 任務執行結束之后，將任務結果保存在數據庫中: 

 

 

 5.3定時任務

如果我們想某日某時執行某個任務，或者每隔一段時間執行某個任務，也可以使用celery來完成.   使用定時任務，需要安裝額外包:

pip install django_celery_beat
  首先在settings.py中安裝此應用:

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'demo',
    'django_celery_results',
    'django_celery_beat',  # 安裝應用
]
  在celery_demo/celery.py模塊中增加定時任務配置:

from celery import Celery
from django.conf import settings
import os

# 為celery設置環境變量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_demo.settings')

# 創建應用
app = Celery("demo")
# 配置應用
app.conf.update(
    # 配置broker, 這里我們用redis作為broker
    BROKER_URL='redis://:332572@127.0.0.1:6379/1',
    # 使用項目數據庫存儲任務執行結果
    CELERY_RESULT_BACKEND='django-db',
    # 配置定時器模塊，定時器信息存儲在數據庫中
    CELERYBEAT_SCHEDULER='django_celery_beat.schedulers.DatabaseScheduler',

)
# 設置app自動加載任務
# 從已經安裝的app中查找任務
app.autodiscover_tasks(settings.INSTALLED_APPS)
  由於定時器信息存儲在數據庫中，我們需要先生成對應表, 對diango_celery_beat執行遷移操作，創建對應表:

python manage.py migrate django_celery_beat

 

  我們可登錄網站后台Admin去創建對應任務, 首先我們先在tasks.py模塊中增加新的任務，用於定時去執行(5秒執行一次)

from celery_demo.celery import app
import time

# 用於定時執行的任務
@app.task
def interval_task():
    print("我每隔5秒鍾時間執行一次....")
  首先創建后台管理員帳號:

python manage.py createsuperuser
  登錄管理后台Admin:

 

  其中Crontabs用於定時某個具體時間執行某個任務的時間，Intervals用於每隔多久執行任務的事件，具體任務的執行在Periodic tasks表中創建。

  我們要創建每隔5秒執行某個任務，所以在Intervals表名后面點擊Add按鈕:

 

  然后在Periodic tasks表名后面，點擊Add按鈕，添加任務:

 

  啟動定時任務：

celery -A celery_demo worker -l info --beat

 

  任務每隔5秒中就會執行一次，如果配置了存儲，那么每次任務執行的結果也會被保存到對應的數據庫中。