將celery集成到Django項目中,，實現(xiàn)異步任務處理和定時任務處理

1 Celery安裝與配置

在虛擬環(huán)境中安裝:

pip install django-celery==3.2.2

pip install redis

pip install flower

查看集成到Django中的celery版本,， pip freeze

celery==3.1.26.post2 django-celery==3.2.2 flower==0.9.2

啟動redis服務,， 端口假設為6379

發(fā)現(xiàn)pip安裝比較慢的情況

pip install pillow -i https://pypi.douban.com/simple

2 Django中配置

（1）在主工程的配置文件settings.py 中應用注冊表INSTALLED_APPS中加入 djcelery

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'art',
    'xadmin',
    'crispy_forms',
    'DjangoUeditor',
    'djcelery',       #加入djcelery
]

(2) 在settings.py 中加入celery配置信息

#############################
# celery 配置信息 start
#############################
import djcelery
djcelery.setup_loader()
BROKER_URL = 'redis://127.0.0.1:6379/1'
CELERY_IMPORTS = ('art.tasks')
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' 
from celery.schedules import crontab
from celery.schedules import timedelta
?
CELERYBEAT_SCHEDULE = {    #定時器策略
    #定時任務一：　每隔30s運行一次
    u'測試定時器1': {
        "task": "art.tasks.tsend_email",
        #"schedule": crontab(minute='*/2'),  # or 'schedule':   timedelta(seconds=3),
        "schedule":timedelta(seconds=30),
        "args": (),
    },
}
#############################
# celery 配置信息 end
#############################

? 當djcelery.setup_loader()運行時,，Celery便會去查看INSTALLD_APPS下包含的所有app目錄中的tasks.py文件,，找到標記為task的方法,，將它們注冊為celery task

? BROKER_URL：broker是代理人,，它負責分發(fā)任務給worker去執(zhí)行。我使用的是Redis作為broker

? 沒有設置 CELERY_RESULT_BACKEND,，默認沒有配置,，此時Django會使用默認的數(shù)據(jù)庫(也是你指定的orm數(shù)據(jù)庫)。

CELERY_IMPORTS：是導入目標任務文件

CELERYBEAT_SCHEDULER：使用了django-celery默認的數(shù)據(jù)庫調度模型,任務執(zhí)行周期都被存在默認指定的orm數(shù)據(jù)庫中．

CELERYBEAT_SCHEDULE：設置定時的時間配置,， 可以精確到秒,，分鐘，小時,，天,，周等。

（3）創(chuàng)建應用實例

? 在主工程目錄添加celery.py,， 添加自動檢索django工程tasks任務

? vim artproject/celery.py

#!/usr/bin/env python  
# encoding: utf-8  
#目的是拒絕隱士引入,，celery.py和celery沖突。
from __future__ import absolute_import,unicode_literals 
import os
from celery import Celery
from django.conf import settings
?
?
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "artproject.settings")
?
#創(chuàng)建celery應用
app = Celery('art_project')
#You can pass the object directly here, but using a string is better since then the worker doesn’t have to serialize the object.
app.config_from_object('django.conf:settings')
#如果在工程的應用中創(chuàng)建了tasks.py模塊,，那么Celery應用就會自動去檢索創(chuàng)建的任務,。比如你添加了一個任#務,，在django中會實時地檢索出來。
app.autodiscover_tasks(lambda :settings.INSTALLED_APPS)

(4) 創(chuàng)建任務 tasks

每個任務本質上就是一個函數(shù),，在tasks.py中,，寫入你想要執(zhí)行的函數(shù)即可。

在應用art中添加我們需要提供的異步服務和定時服務

vim art/tasks.py

#!/usr/bin/env python  
# encoding: utf-8  
from __future__ import absolute_import
import time
from django.core.mail import send_mail
from celery.utils.log import get_task_logger
from artproject.celery import app
?
from art.utils.send_mail import pack_html, send_email
?
@app.task
def tsend_email():
   url = "http://"
   receiver = 'zhouguangyou@'
   content = pack_html(receiver, url)
   # content = 'this is email content.'
   send_email(receiver, content)
   print('send email ok!')
?
?
@app.task
def add(x, y):
   return x+y
?

上述我們把異步處理任務add和定時器任務tsend_email都放在了tasks.py 中

（5）遷移生成celery需要的數(shù)據(jù)表

python manage.py migrate

此時數(shù)據(jù)庫表結構多出了幾個

celery_taskmeta            |
| celery_tasksetmeta         |
| djcelery_crontabschedule   |
| djcelery_intervalschedule  |
| djcelery_periodictask      |
| djcelery_periodictasks     |
| djcelery_taskstate         |
| djcelery_workerstate 

3 啟動服務,，測試

我們可以采用 python manage.py help 發(fā)現(xiàn)多出了 celery 相關選項,。

（1）啟動django celery 服務

啟動服務：

python manage.py celery worker --loglevel=info

此時異步處理和定時處理服務都已經(jīng)啟動了

（2）web端接口觸發(fā)異步任務處理

我們在web端加入一個入口，觸發(fā)異步任務處理add函數(shù)

在應用art的urls.py 中加入如下對應關系

from art.views import add_handler
?
?
url(r'^add', add_handler),

art/views.py 中加入處理邏輯

def add_handler(request):
   x = request.GET.get('x', '1')
   y = request.GET.get('y', '1')
   from .tasks import add
   add.delay(int(x), int(y))
   res = {'code':200, 'message':'ok', 'data':[{'x':x, 'y':y}]}
   return HttpResponse(json.dumps(res))

啟動web服務,，通過url傳入的參數(shù),，通過handler的add.delay(x, y)計算并存入mysql

http://127.0.0.1:8000/art/add?x=188&y=22

(4) 測試定時器，發(fā)送郵件

在終端輸入 python manage.py celerybeat -l info

會自動觸發(fā)每隔30s執(zhí)行一次tsend_email定時器函數(shù),，發(fā)送郵件：

CELERYBEAT_SCHEDULE = {    #定時器策略
    #定時任務一：　每隔30s運行一次
    u'測試定時器1': {
        "task": "art.tasks.tsend_email",
        #"schedule": crontab(minute='*/2'),  # or 'schedule': timedelta(seconds=3),
        "schedule":timedelta(seconds=30),
        "args": (),
    },
}

具體發(fā)送郵件服務程序見下面的第4節(jié)

4 郵件發(fā)送服務

項目中經(jīng)常會有定時發(fā)送郵件的情形,，比如發(fā)送數(shù)據(jù)報告，發(fā)送異常服務報告等,。

可以編輯文件 art/utils/send_mail.py, 內容編輯如下：

#!/usr/bin/env python
#-*- coding:utf-8 -*-
#written by zhouguangyou
#發(fā)送郵件(wd_email_check123賬號用于內部測試使用,，不要用于其他用途)
?
import smtplib  
from email.mime.multipart import MIMEMultipart  
from email.mime.text import MIMEText  
from email.mime.image import MIMEImage 
from email.header import Header
import time
?
sender = '[email protected]'  
subject = u'api開放平臺郵箱驗證'
smtpserver = 'smtp.163.com'
username = 'wd_email_check123'
password = 'wandacheck1234'
mail_postfix="163.com"
?
def send_email(receiver, content):
    try:
        me = username+"<"+username+"@"+mail_postfix+">"
        msg = MIMEText(content, 'html', 'utf-8')
        msg['Subject'] = subject
        msg['From'] = sender
        msg['To'] = receiver
        smtp = smtplib.SMTP()  
        smtp.connect(smtpserver)  
        smtp.login(username, password)
        smtp.sendmail(sender, receiver, msg.as_string())  
        smtp.quit()
        return True
    except Exception as e:
        print('send_email has error with : ' + str(e))
        return False
?
?
def pack_html(receiver, url):
    html_content = u"<html><div>尊敬的用戶<font color='#0066FF'>%s</font> 您好！</div><br>" \
                   "<div>感謝您關注我們的平臺 ,，我們將為您提供最貼心的服務,，祝您購物愉快。</div><br>" \
                   "<div>點擊以下鏈接,，即可完成郵箱安全驗證：</div><br>"  \
                   "<div><a href='%s'>%s</a></div><br>"  \
                   "<div>為保障您的賬號安全,，請在24小時內點擊該鏈接; </div><br>" \
                   "<div>若您沒有申請過驗證郵箱 ，請您忽略此郵件,，由此給您帶來的不便請諒解,。</div>" \
                   "</html>" % (receiver, url, url)
    html_content = html_content
    return html_content
?
?
if __name__ == "__main__":
    url = "http://"
    receiver = 'zhouguangyou@'
    #content = pack_html(receiver, url)
    content = 'this is email content. at %s.'%int(time.time())
    send_email(receiver,  content)

至此，在celery ui界面可以看到兩類,，定時器處理和異步處理,。