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数字表述更明显
CELERY_ACCEPT_CONTENT = ['json'] # 指定接受的内容类型
CELERYBEAT_SCHEDULE = {
'send_mail': {
'task': 'celery_test.tasks.send_mail',
'schedule': timedelta(seconds=10),
}
}
2、函数模块
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","第三方发送验证码")
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,执行任务
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}
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。