許多Django應用需要執行非同步任務, 以便不耽誤http request的執行. 我們也可以選擇許多方法來完成非同步任務, 使用Celery是一個比較好的選擇, 因為Celery有著大量的社群支援, 能夠完美的擴充套件, 和Django結合的也很好. Celery不僅能在Django中使用, 還能在其他地方被大量的使用. 因此一旦學會使用Celery, 我們可以很方便的在其他專案中使用它.

Celery介紹

• Celery的主要用處是執行非同步任務, 可以選擇延期或定時執行功能
• 適用於以下的場景
• 第一, 假設使用者正發起一個request, 並等待request完成後返回. 在這一request後面的view功能中, 我們可能需要執行一段花費很長時間的程式任務, 這一時間可能遠遠大於使用者能忍受的範圍. 當這一任務並不需要立刻執行時, 我們便可以使用Celery在後臺執行, 而不影響使用者瀏覽網頁. 當有任務需要訪問遠端伺服器完成時, 我們往往都無法確定需要花費的時間.
• 第二則是定期執行某些任務. 比如每小時需要檢查一下天氣預報, 然後將資料儲存到資料庫中. 我們可以編寫這一任務, 然後讓Celery每小時執行一次. 這樣我們的web應用便能獲取最新的天氣預報資訊.

我們這裡所講的任務task, 就是一個Python功能(function). 定期執行一個任務可以被認為是延時執行該功能. 我們可以使用Celery延遲5分鐘呼叫function task1, 並傳入引數(1, 2, 3). 或者我們也可以每天午夜執行該function.

我們偏向於將Celery放入專案中, 便於task訪問統一資料庫和Django設定.

當task準備執行時, Celery會將其放入列隊queue中. queue中儲存著可以執行的task的list. 我們可以使用多個queue,

將任務task放入queue就像加入todo list一樣. 為了使task執行, 我們還需要在其他執行緒中執行的苦工worker. worker實時觀察著代執行的task, 並逐一執行這些task. 你可以使用多個worker, 通常他們位於不同伺服器上.

安裝Celery

pip install django-celery

django 中的配置

暫時使用django runserver來啟動celery. 而Celery代理人(broker), 我們使用Django database broker implementation. 現在我們只需要知道Celery需要broker, 使用django自身便可以充當broker. (但在部署時, 我們最好使用更穩定和高效的broker, 例如Redis.)

在settings中配置

import djcelery
    djcelery.setup_loader()
    BROKER_URL = 'django://'
    ...
    INSTALLED_APPS = (
       ...
       'djcelery',
       'kombu.transport.django',
       ...
    )

第一二項是必須的, 第三項則告訴Celery使用Django專案作為broker.

在INSTALLED_APPS中新增的djcelery是必須的. kombu.transport.django則是基於Django的broker

最後建立Celery所需的資料表, 使用makemigrations 和 migrate 命令建立表結構

建立一個task任務

正如前面所說的, 一個task就是一個Pyhton function. 但Celery需要知道這一function是task, 因此我們可以使用celery自帶的裝飾器decorator: @task. 在django app目錄中建立taske.py:

 from celery import task

    @task()
    def add(x, y):
        return x + y

當settings.py中的djcelery.setup_loader()執行時, Celery便會檢視所有INSTALLED_APPS中app目錄中的tasks.py檔案, 找到標記為task的function, 並將它們註冊為celery task.

將function標註為task並不會妨礙他們的正常執行. 你還是可以像平時那樣呼叫它: z = add(1, 2).

開始執行task

讓我們以一個簡單的例子作為開始. 例如我們希望在使用者發出request後非同步執行該task, 馬上返回response, 從而不阻塞該request, 使使用者有一個流暢的訪問過程. 那麼, 我們可以使用.delay, 例如在在views.py的一個view中:

  from myapp.tasks import add
    ...
        add.delay(2, 2)
    ...

Celery會將task加入到queue中, 並馬上返回. 而在一旁待命的worker看到該task後, 便會按照設定執行它, 並將他從queue中移除. 而worker則會執行以下程式碼:

    import myapp.tasks.add

    myapp.tasks.add(2, 2)

關於 import

這裡需要注意的是, 在impprt task時, 需要保持一致. 因為在執行djcelery.setup_loader()時, task是以INSTALLED_APPS中的app名, 加.tasks.function_name註冊的, 如果我們由於python path不同而使用不同的引用方式時(例如在tasks.py中使用from myproject.myapp.tasks import add形式), Celery將無法得知這是同一task, 因此可能會引起奇怪的bug.

測試

• 啟動worker
  正如之前說到的, 我們需要worker來執行task. 以下是在開發環境中的如何啟動worker:

首先啟動terminal, 如同開發django專案一樣, 啟用virtualenv, 切換到django專案目錄. 然後啟動django自帶web伺服器: python manage.py runserver.

然後啟動worker:

python manage.py celery worker --loglevel=info

此時, worker將會在該terminal中執行, 並顯示輸出結果.

• 啟動task
• 開啟新的terminal, 啟用virtualenv, 並切換到django專案目錄:

 $ python manage.py shell
    >>> from myapp.tasks import add
    >>> add.delay(2, 2)

此時, 你可以在worker視窗中看到worker執行該task:

下面我們來看一個例子, 在views.py和tasks.py中:

# views.py
    from myapp.tasks import do_something_with_form_data

    def view(request):
        form = SomeForm(request.POST)
        if form.is_valid():
            data = form.cleaned_data
            # Schedule a task to process the data later
            do_something_with_form_data.delay(data)
        return render_to_response(...)

# tasks.py

    @task
    def do_something_with_form_data(data):
        call_slow_web_service(data['user'], data['text'], ...)

除錯

由於Celery的執行需要啟動多個部件, 我們可能會漏掉一兩個. 所以我們建議:

使用最簡單的設定
使用python debug和logging功能顯示當前的程序

Eager模式

如果在settings.py設定:

    CELERY_ALWAYS_EAGER = True
那麼Celery便以eager模式執行, 則task便不需要加delay執行:

    # 若啟用eager模式, 則以下兩行程式碼相同
    add.delay(2, 2)
    add(2, 2)

檢視queue

因為我們使用了django作為broker, queue儲存在django的資料庫中. 這就意味著我們可以通過django admin檢視該queue:

# admin.py
    from django.contrib import admin
    from kombu.transport.django import models as kombu_models

    admin.site.register(kombu_models.Message)

檢查結果

每次執行非同步task後, Celery都會返回AsyncResult物件作為結果. 你可以將其儲存, 然後在將來檢視該task是否執行成功和返回結果:

# views.py

result = add.delay(2, 2)
...
if result.ready():
    print "Task has run"
    if result.successful():
        print "Result was: %s" % result.result
    else:
        if isinstance(result.result, Exception):
            print "Task failed due to raising an exception"
            raise result.result
        else:
            print "Task failed without raising exception"
 else:
     print "Task has not yet run"

定期任務

還有一種Celery的常用模式便是執行定期任務. 執行定期任務時, Celery會通過celerybeat程序來完成. Celerybeat會保持執行, 一旦到了某一定期任務需要執行時, Celerybeat便將其加入到queue中. 不像worker程序, Celerybeat只有需要一個即可.

啟動Celerybeat:

python manage.py celery beat

使Celery執行定期任務的方式有很多種, 我們先看第一種, 將定期任務儲存在django資料庫中. 即使是在django和celery都執行的狀態, 這一方式也可以讓我們方便的修改定期任務. 我們只需要設定settings.py中的一項便能開啟這一方式:

# settings.py
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'

然後我們便可以通過django admin的/admin/djcelery/periodictask/新增定期任務了.

Name: 這一定期任務的註冊名
Task (registered): 可以選擇所有已經註冊的task之一, 例如前面的add function
Task (custom): task的全名, 例如myapp.tasks.add, 但最好還是用以上項
Enabled: 是否開啟這一定期任務
Interval: 定期任務的間隔時間, 例如每隔5分鐘
Crontab: 如果希望task在某一特定時間執行, 則使用Unix中的Crontab代替interval
Arguments: 用於傳引數到task中