celery例子實現(xiàn)定時發(fā)送郵件

1、配置文件

vi setting.py

#coding:utf-8
import?os
from?datetime?import??timedelta
BROKER_URL?=?'redis://127.0.0.1:6379/13'
CELERY_RESULT_BACKEND?=?'redis://127.0.0.1:6379/14'
CELERY_TIMEZONE?=?'Asia/Shanghai'
CELERY_ENABLE_UTC?=?True
CELERY_TASK_SERIALIZER?=?'json'?#?任務序列化和反序列化使用msgpack方案
#
CELERY_RESULT_SERIALIZER?=?'json'?#?讀取任務結果一般性能要求不高，所以使用了可讀性更好的JSON
#
CELERY_TASK_RESULT_EXPIRES?=?60?*?60?*?24?#?任務過期時間，不建議直接寫86400，應該讓這樣的magic數(shù)字表述更明顯

CELERY_ACCEPT_CONTENT?=?['json']?#?指定接受的內(nèi)容類型
CELERYBEAT_SCHEDULE?=?{

????'send_mail':?{

????????'task':?'celery_test.tasks.send_mail',

???????'schedule':?timedelta(seconds=10),

????}
}

2、函數(shù)模塊

vi tasks.py

#coding:utf-8
from?server?import?app
import?random,string,smtplib
@app.task
def?send_mail():
????????SUBJECT="臨時登錄密碼"
????????HOST="smtp.163.com"
????????#?TO=mail
????????TO='456@qq.com'
????????FROM="123@163.com"
????????text=str(random.randint(1000,9999))
????????BODY=string.join((

????????"From:%s"%FROM,

????????"To:%s"%TO,

????????"Subject:%s"%SUBJECT,

????????"",text),"\r\n")

????????server=smtplib.SMTP(HOST)

????????server.login("123@qq.com","第三方發(fā)送驗證碼")

????????server.sendmail(FROM,[TO],BODY)
????????server.quit()
????????return?True

3、啟動模塊

vi server.py

from??celery?import?Celery
app=Celery('celery_test',include=['celery_test.tasks'])
app.config_from_object('celery_test.setting')

if?__name__=='__main__':
????app.start()

4、啟動worker，執(zhí)行任務

celery -A celery_test.server worker -l info

5、啟動beat，生成任務

celery -A celery_test.server beat

6、目錄結構

celery_test/{server.py,setting.py,tasks.py}