Celery由以下三部分构成:消息中间件(Broker)、任务执行单元Worker、结果存储(Backend)
工作原理:
消息中间件Broker
消息中间件Broker官方提供了很多备选方案,支持RabbitMQ、Redis、Amazon SQS、MongoDB、Memcached 等,官方推荐RabbitMQ。
任务执行单元Worker
Worker是任务执行单元,负责从消息队列中取出任务执行,它可以启动一个或者多个,也可以启动在不同的机器节点,这就是其实现分布式的核心。
结果存储Backend
Backend结果存储官方也提供了诸多的存储方式支持:RabbitMQ、 Redis、Memcached,SQLAlchemy, Django ORM、Apache Cassandra、Elasticsearch。
pip install selery #消息队列异步发送
pip install ronglian_sms_sdk #短信平台sdk
主文件:创建实例,加载配置文件,检测任务
"""
Celery 将这三者串联起来
生产者 队列 消费者
1. 创建celery
2. 设置队列(broker)
3. 设置生产者(任务 task)
① 任务的本质就是函数
② 这个函数必须要被celery的实例对象的 task装饰器装饰
③ 必须调用celery实例对象的自动检测来检测任务
4. 设置消费者(worker)
celery -A celery实例对象的文件 worker -l info -p eventlet
celery -A celery_tasks.main worker -l info -P gevent
"""
#① 让celery去加载我们当前工程中的配置文件
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "meiduo_mall.settings")
#② 创建celery实例对象
from celery import Celery
# celery的第一个参数是main
# 习惯上,填写当前脚本的工程名就可以
# 给celery的实例起个名字,这个名字唯一就可以
app = Celery('celery_tasks')
#③ celery 设置 broker (队列)
# config_from_object 参数: 就是 配置文件的路径
app.config_from_object('celery_tasks.config')
#④ 让celery自动检测任务
#autodiscover_tasks 参数是 列表
# 列表的元素是: 任务的包路径
app.autodiscover_tasks(['celery_tasks.sms'])
配置文件
##!/usr/bin/env python3
## -*- coding:utf-8 -*-
## Author:wd
#from project import app
#from celery.schedules import crontab
#BROKER_URL = 'redis://10.1.210.69:6379/0' # Broker配置,使用Redis作为消息中间件
#CELERY_RESULT_BACKEND = 'redis://10.1.210.69:6379/0' # BACKEND配置,这里使用redis
#CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
#CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间
#CELERY_TIMEZONE='Asia/Shanghai' # 时区配置
#CELERY_IMPORTS = ( # 指定导入的任务模块,可以指定多个
# 'project.tasks',
# 'project.period_task',
#)
#app.conf.beat_schedule = {
# 'period_add_task': { # 计划任务
# 'task': 'project.period_task.add', #任务路径
# 'schedule': crontab(hour=18, minute=16, day_of_week=1),
# 'args': (3, 4),
# },
#'add-every-30-seconds': { # 每10秒执行
# 'task': 'project.period_task.sayhi', #任务路径
# 'schedule': 10.0,
# 'args': ('wd',)
# },
#}
# 我们选择的是redis作为我们的 队列
# 选择redis的 14号库
broker_url = "amqp://guest@localhost//"
#broker_url = "redis://127.0.0.1/14"
# 结果选择15号库
result_backend = "redis://127.0.0.1/15"
任务模块
"""
① 任务的本质就是函数
② 这个函数必须要被celery的实例对象的 task装饰器装饰
③ 必须调用celery实例对象的自动检测来检测任务
"""
from celery_tasks.main import app
tid = 1 #默认模板
accId = ''#填写自己的
accToken = ''#填写自己的
appId = ''#填写自己的
from ronglian_sms_sdk import SmsSDK
@app.task
def send_sms_code(mobile,sms_code):
sdk = SmsSDK(accId, accToken, appId)
datas = (str(sms_code),'2')
resp = sdk.sendMessage(str(tid), str(mobile), datas)
启动celery
celery -A celery_tasks.main worker -l info -P gevent
#发送短信验证码
class SmsCodeView(View):
def get(self,request,mobile):
# 1. 接收参数(手机号,图片验证码,uuid)
image_code = request.GET.get('image_code')
uuid=request.GET.get('image_code_id')
# 2. 验证参数
if not all([mobile,image_code,uuid]):
# 响应码
return http.JsonResponse({'code':RETCODE.NECESSARYPARAMERR,'errormsg':'参数不齐'})
# 3. 验证用户输入的图片验证码和服务器保存的图片验证码一致
try:
redis_conn = get_redis_connection('code')
redis_code = redis_conn.get('img_%s'%uuid)
if redis_code is None:
return http.JsonResponse({'code':RETCODE.IMAGECODEERR,'errmsg':'图片验证码过期'})
# 添加一个删除图片验证码的逻辑
redis_conn.delete('img_%s'%uuid)
except Exception as e:
logger.error(e)
return http.JsonResponse({'code':RETCODE.DBERR,'errmsg':'redis有异常'})
# 我们获取redis的数据都是bytes类型
if redis_code.decode().lower() != image_code.lower():
return http.JsonResponse({'code':RETCODE.SMSCODERR,'errmsg':'图片验证码错误'})
# 判断标记位是否为1
send_flag = redis_conn.get('send_flag_%s'%mobile)
if send_flag:
return http.JsonResponse({'code':RETCODE.THROTTLINGERR,'errmsg':'操作太频繁'})
# 4. 先生成一个随机短信码
from random import randint
# 不满6位 补齐
sms_code = '%06d'%randint(0,999999)
# 5. 先把短信验证码保存起来
pipe = redis_conn.pipeline()
pipe.setex('sms_%s' % mobile, SMS_CODE_EXPIRE_TIME, sms_code)
pipe.setex('send_flag_%s' % mobile, 60, 1)
#让管道执行
pipe.execute()
# 6. 最后发送
# 我们的函数 需要通过delay调用 才能添加到 broker(队列)中
from celery_tasks.sms.tasks import send_sms_code
## send_sms_code 的参数 平移到 delay中
send_sms_code.delay(mobile,sms_code)
#sdk = SmsSDK(accId, accToken, appId)
#datas = (str(sms_code),'2')
#resp = sdk.sendMessage(str(tid), str(mobile), datas)
return http.JsonResponse({'code':RETCODE.OK,'errmsg':'ok'})
celery的作用
对于(钩子函数,监控)等不在多做阐述
参考链接 celery用法:https://www.cnblogs.com/wdliu/p/9517535.html django中celery用法:https://www.cnblogs.com/wdliu/p/9530219.html