基於Django與Celery實現非同步佇列任務

在運營系統中經常用到非同步方式來處理我們的任務，比如將業務上線流程串成任務再寫入佇列，通過後臺作業節點去排程執行。比較典型的案例為騰訊的藍鯨、織雲、雲智慧等平臺。本譯文結合Django+Celery+Redis實現一個定期從Flickr 獲取圖片並展示的簡單案例，方便大家理解實現非同步對列任務的過程。

剛接觸django的時候，我經歷過的最讓人沮喪的事情是需要定期執行一段程式碼。我寫了一個需要每天上午12點執行一個動作的不錯的函式。很簡單是不是？錯了。事實證明，這對我來說是一個巨大的困難點，因為，那時我使用Cpane型別的虛擬主機管理系統，它能專門提供一個很友好，很方便的圖形使用者介面來設定cron作業。

經過反覆研究，我發現了一個很好的解決方案 – Celery，一個用於在後臺執行任務的強大的非同步作業佇列。但是，這也導致了其它的問題，因為我無法找到一系列簡單的指令將celery整合到Django專案中。

當然，我最終還是設法成功搞定了它 – 這正是本文將介紹的內容：如何將celery整合到一個Django專案，建立週期性任務。

該專案利用Python3.4，Django的1.8.2，celery3.1.18和Redis3.0.2.

一、概述

由於大篇幅的文字，為了您的方便，請參閱下表中的每一步的簡要資訊，並獲取相關的程式碼。

步驟概要 Git標籤
樣板樣板下載

V1
建立整合Celery和Django V2
Celery任務新增基本的Celery任務   V3
週期性任務新增週期性任務   V4
本地執行本地執行我們的應用程式   V5
遠端執行遠端執行我們的應用程式   V5

二、什麼是Celery

“Celery是一個非同步任務佇列/基於分散式訊息傳遞的作業佇列。它側重於實時操作，但對排程的支援也很好。”本文，我們將重點講解週期性執行任務的排程特點。

為什麼這一點有用呢？

回想一下你不得不在將來執行某一特定任務的經歷。也許你需要每隔一小時訪問一個API。或者，也許你需要在這一天結束時傳送一批電子郵件。不論任務大小，Celery都可以使得排程週期性任務變的很容易。

你永遠不希望終端使用者等待那些不必要的頁面載入或動作執行完成。如果你的應用程式工作流的一部分是一個需要很長時間的程式，當資源可用時，你就可以使用Celery在後臺執行這段程式，從而使你的應用程式可以繼續響應客戶端的請求。這樣可以使任務在應用程式的環境之外執行。

三、構建專案

在深入瞭解Celery之前，先從Github庫中獲取開始專案。確保啟用一個虛擬的環境，安裝必要的軟體，並執行遷移。然後啟動伺服器，通過你的瀏覽器導航到http://localhost:8000/。你應當能看到‘恭喜你的第一個Django頁面’。完成後，關閉伺服器。
接下來，我們開始安裝celery。

Python

$ pip install celery==3.1.18 
$ pip freeze > requirements.txt

12	$pip install celery==3.1.18$pip freeze>requirements.txt

現在，我們通過簡單的三步將celery整合到django專案中。

步驟一：建立celery.py

在“picha“目錄下，建立celery.py，程式碼如下：

Python

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

# set the default Django settings module for the 'celery' program. 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'picha.settings') 
app = Celery('picha') 

# Using a string here means the worker will not have to 
# pickle the object when using Windows. 
app.config_from_object('django.conf:settings') 
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) 

@app.task(bind=True) 
def debug_task(self): 
    print('Request: {0!r}'.format(self.request))

1234567891011121314151617

from__future__importabsolute_import importosfromcelery importCelery fromdjango.conf importsettings# set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE','picha.settings')app=Celery('picha')# Using a string here means the worker will not have to # pickle the object when using Windows. app.config_from_object('django.conf:settings')app.autodiscover_tasks(lambda:settings.INSTALLED_APPS)@app.task(bind=True)defdebug_task(self):print('Request: {0!r}'.format(self.request))

請注意程式碼中的註釋。

步驟二：引入celery應用

為了確保在django啟動時載入了celery應用，在settings.py旁邊新建__init__.py，並新增以下程式碼到__init__.py中。

Python

from __future__ import absolute_import 

# This will make sure the app is always imported when 
# Django starts so that shared_task will use this app. 
from .celery import app as celery_app

12345

from__future__importabsolute_import# This will make sure the app is always imported when # Django starts so that shared_task will use this app. from.celery importapp ascelery_app

完成以上步驟後，你的專案目錄應該是這樣的：

Python

├── manage.py 
├── picha 
│   ├── __init__.py 
│   ├── celery.py 
│   ├── settings.py 
│   ├── urls.py 
│   └── wsgi.py 
└── requirements.txt

12345678

├──manage.py├──picha│├──__init__.py│├──celery.py│├──settings.py│├──urls.py│└──wsgi.py└──requirements.txt

步驟三：安裝 Redis作為Celery的“中介軟體”

Celery使用中介軟體在django專案與celery監控者之間傳遞訊息。在本教程中，我們使用redis作為訊息中間代理。

首先，從官方下載頁面或通過brew（BREW安裝Redis）安裝Redis，然後開啟你的終端上，在一個新的終端視窗，啟動伺服器：

Python

$ redis-server

1	$redis-server

你可以通過在終端中輸入如下命令測試Redis是否正常工作。

Python

$ redis-cli ping

1	$redis-cli ping

Redis應該回復PONG – 試試吧！
一旦Redis正常啟動了，把下面的程式碼新增到你的settings.py檔案中：

Python

# CELERY STUFF 
BROKER_URL = 'redis://localhost:6379' 
CELERY_RESULT_BACKEND = 'redis://localhost:6379' 
CELERY_ACCEPT_CONTENT = ['application/json'] 
CELERY_TASK_SERIALIZER = 'json' 
CELERY_RESULT_SERIALIZER = 'json' 
CELERY_TIMEZONE = 'Africa/Nairobi'

1234567

# CELERY STUFF BROKER_URL='redis://localhost:6379'CELERY_RESULT_BACKEND='redis://localhost:6379'CELERY_ACCEPT_CONTENT=['application/json']CELERY_TASK_SERIALIZER='json'CELERY_RESULT_SERIALIZER='json'CELERY_TIMEZONE='Africa/Nairobi'

你還需要新增Redis的作為Django專案的依賴：

Python

$ pip install redis==2.10.3 
$ pip freeze > requirements.txt

12	$pip install redis==2.10.3$pip freeze>requirements.txt

就是這樣了！你現在應該能夠在Django中使用Celery。有關設定Celery與Django的更多資訊，請檢視官方Celery文件。

在繼續下面步驟之前，讓我們進行一些完整性檢查，以確保一切都是正常的。

測試Celery worker已準備好接收任務：

Python

$ celery -A picha worker -l info 
... 
[2015-07-07 14:07:07,398: INFO/MainProcess] Connected to redis://localhost:6379// 
[2015-07-07 14:07:07,410: INFO/MainProcess] mingle: searching for neighbors 
[2015-07-07 14:07:08,419: INFO/MainProcess] mingle: all alone

12345

$celery-Apicha worker-linfo...[2015-07-0714:07:07,398:INFO/MainProcess]Connected to redis://localhost:6379//[2015-07-0714:07:07,410:INFO/MainProcess]mingle:searching forneighbors[2015-07-0714:07:08,419:INFO/MainProcess]mingle:allalone

使用CTRL-C殺死該段程式。現在，測試Celery任務排程程式是否已經準備好：

Python

$ celery -A picha beat -l info 
... 
[2015-07-07 14:08:23,054: INFO/MainProcess] beat: Starting...

123	$celery-Apicha beat-linfo...[2015-07-0714:08:23,054:INFO/MainProcess]beat:Starting...

在上述完成時再次終止該程序。

1、Celery任務

Celery利用celery呼叫的常規Python函式作為任務。

例如，讓我們把這個基本函式變為celery的任務：

Python

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

12	defadd(x,y):returnx+y

首先，新增一個裝飾器。

Python

from celery.decorators import task 

@task(name="sum_two_numbers") 
def add(x, y): 
    return x + y

12345

fromcelery.decorators importtask@task(name="sum_two_numbers")defadd(x,y):returnx+y

然後你可以通過以下方式利用celery非同步執行該任務：

Python

add.delay(7, 8)

1	add.delay(7,8)

很簡單，對不對？

所以，這對於解決類似你要載入一個網頁，而不需要使用者等待一些後臺程式的完成這些型別的任務來說是非常完美的。

讓我們來看一個例子…

讓我們再回到Django專案的版本3，它包括一個接受來自使用者的反饋的應用程式，人們形象地稱之為反饋：

Python

├── feedback  
│   ├── __init__.py  
│   ├── admin.py  
│   ├── emails.py  
│   ├── forms.py  
│   ├── models.py  
│   ├── tests.py  
│   └── views.py  
├── manage.py  
├── picha  
│   ├── __init__.py  
│   ├── celery.py  
│   ├── settings.py  
│   ├── urls.py  
│   └── wsgi.py  
├── requirements.txt  
└── templates  
    ├── base.html  
    └── feedback  
        ├── contact.html  
        └── email  
            ├── feedback_email_body.txt  
            └── feedback_email_subject.txt

1234567891011121314151617181920212223

├──feedback│├──__init__.py│├──admin.py│├──emails.py│├──forms.py│├──models.py│├──tests.py│└──views.py├──manage.py├──picha│├──__init__.py│├──celery.py│├──settings.py│├──urls.py│└──wsgi.py├──requirements.txt└──templates├──base.html└──feedback├──contact.html└──email├──feedback_email_body.txt└──feedback_email_subject.txt

安裝新的必要軟體，啟動應用程式，並導航到http://localhost:8000/feedback/。你應該看到如下結果：

讓我們連線celery任務。

2、新增任務

基本上，使用者提交反饋表後，我們希望讓他繼續以他舒服的方式往下進行，而我們在後臺進行處理反饋，傳送電子郵件等等。

要做到這一點，首先新增一個叫tasks.py的檔案到“feedback”目錄：

Python

from celery.decorators import task  
from celery.utils.log import get_task_logger  

from feedback.emails import send_feedback_email  

logger = get_task_logger(__name__)  

@task(name="send_feedback_email_task")  
def send_feedback_email_task(email, message):  
    """sends an email when feedback form is filled successfully"""  
    logger.info("Sent feedback email")  
    return send_feedback_email(email, message)

123456789101112

fromcelery.decorators importtask fromcelery.utils.log importget_task_logger fromfeedback.emails importsend_feedback_email logger=get_task_logger(__name__)@task(name="send_feedback_email_task")defsend_feedback_email_task(email,message):"""sends an email when feedback form is filled successfully"""logger.info("Sent feedback email")returnsend_feedback_email(email,message)

然後按照如下內容更新forms.py：

Python

from django import forms  
from feedback.tasks import send_feedback_email_task  

class FeedbackForm(forms.Form):  
    email = forms.EmailField(label="Email Address")  
    message = forms.CharField(  
        label="Message", widget=forms.Textarea(attrs={'rows': 5}))  
    honeypot = forms.CharField(widget=forms.HiddenInput(), required=False)  

    def send_email(self):  
        # try to trick spammers by checking whether the honeypot field is  
        # filled in; not super complicated/effective but it works  
        if self.cleaned_data['honeypot']:  
            return False  
        send_feedback_email_task.delay(  
            self.cleaned_data['email'], self.cleaned_data['message'])

12345678910111213141516

基於Django與Celery實現非同步佇列任務

一、概述

二、什麼是Celery

三、構建專案

基於Django與Celery實現非同步佇列任務

Python--Django使用celery實現非同步任務

Django 使用 Celery 實現非同步任務

Django搭配Celery進行非同步/定時任務（一）初步搭建

django與celery結合實現非同步任務

django+celery實現非同步訊息佇列

django —— Celery實現非同步和定時任務

Python Django Celery 實現非同步任務（二）使用rabbitmq 作為broker

celery 與 flask 實現非同步任務排程

java簡單實現非同步佇列:使用生產者與消費者模型

Django框架中，使用celery實現非同步

基於Flask與JS實現任務清單管理

flask中celery介紹及使用celery實現非同步任務

Django配置celery執行非同步任務和定時任務

Celery框架實現非同步執行任務

jQuery基於json與cookie實現購物車的方法

python基於併發與socket實現遠端檔案傳輸程式

django與mysql實現簡單的增刪查改

使用WebAsyncTask與Callable實現非同步Http介面

基於django的webssh實現

基於Django與Celery實現非同步佇列任務

一、概述

二、什麼是Celery

三、構建專案

相關推薦