0.配置文件

# coding:utf-8BROKER_URL = "redis://127.0.0.1:6379/1"
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'

1.启动文件

# coding:utf-8from celery import Celery
from ihome.tasks import config# 定义celery对象
celery_app = Celery("ihome")# 引入配置信息
celery_app.config_from_object(config)# 自动搜寻异步任务
celery_app.autodiscover_tasks(["ihome.tasks.sms"])

2.发送短信辅助类

# coding=utf-8from CCPRestSDK import REST# 主帐号
accountSid = '8aaf0708568d4143015697b0f4960888'# 主帐号Token
accountToken = '42d3191f0e6745d6a9ddc6c795da0bed'# 应用Id
appId = '8aaf0708568d4143015697b0f56e088f'# 请求地址，格式如下，不需要写http://
serverIP = 'app.cloopen.com'# 请求端口
serverPort = '8883'# REST版本号
softVersion = '2013-12-26'# 发送模板短信# @param to 手机号码# @param datas 内容数据 格式为列表 例如：['12','34']，如不需替换请填 ''# @param $tempId 模板Idclass CCP(object):"""自己封装的发送短信的辅助类"""# 用来保存对象的类属性instance = Nonedef __new__(cls):# 判断CCP类有没有已经创建好的对象，如果没有，创建一个对象，并且保存# 如果有，则将保存的对象直接返回if cls.instance is None:obj = super(CCP, cls).__new__(cls)# 初始化REST SDKobj.rest = REST(serverIP, serverPort, softVersion)obj.rest.setAccount(accountSid, accountToken)obj.rest.setAppId(appId)cls.instance = objreturn cls.instancedef send_template_sms(self, to, datas, temp_id):""""""result = self.rest.sendTemplateSMS(to, datas, temp_id)# for k, v in result.iteritems():##     if k == 'templateSMS':#         for k, s in v.iteritems():#             print '%s:%s' % (k, s)#     else:#         print '%s:%s' % (k, v)# smsMessageSid:ff75e0f84f05445ba08efdd0787ad7d0# dateCreated:20171125124726# statusCode:000000status_code = result.get("statusCode")if status_code == "000000":# 表示发送短信成功return 0else:# 发送失败return -1

3.发送短信后端逻辑

# GET /api/v1.0/sms_codes/<mobile>?image_code=xxxx&image_code_id=xxxx
@api.route("/sms_codes/<re(r'1[34578]\d{9}'):mobile>")
def get_sms_code(mobile):"""获取短信验证码"""# 获取参数image_code = request.args.get("image_code")image_code_id = request.args.get("image_code_id")# 校验参数if not all([image_code_id, image_code]):# 表示参数不完整return jsonify(errno=RET.PARAMERR, errmsg="参数不完整")# 业务逻辑处理# 从redis中取出真实的图片验证码try:real_image_code = redis_store.get("image_code_%s" % image_code_id)except Exception as e:current_app.logger.error(e)return jsonify(errno=RET.DBERR, errmsg="redis数据库异常")# 判断图片验证码是否过期if real_image_code is None:# 表示图片验证码没有或者过期return jsonify(errno=RET.NODATA, errmsg="图片验证码失效")# 删除redis中的图片验证码，防止用户使用同一个图片验证码验证多次try:redis_store.delete("image_code_%s" % image_code_id)except Exception as e:current_app.logger.error(e)# 与用户填写的值进行对比if real_image_code.lower() != image_code.lower():# 表示用户填写错误return jsonify(errno=RET.DATAERR, errmsg="图片验证码错误")# 判断对于这个手机号的操作，在60秒内有没有之前的记录，如果有，则认为用户操作频繁，不接受处理try:send_flag = redis_store.get("send_sms_code_%s" % mobile)except Exception as e:current_app.logger.error(e)else:if send_flag is not None:# 表示在60秒内之前有过发送的记录return jsonify(errno=RET.REQERR, errmsg="请求过于频繁，请60秒后重试")# 判断手机号是否存在try:user = User.query.filter_by(mobile=mobile).first()except Exception as e:current_app.logger.error(e)else:if user is not None:# 表示手机号已存在return jsonify(errno=RET.DATAEXIST, errmsg="手机号已存在")# 如果手机号不存在，则生成短信验证码sms_code = "%06d" % random.randint(0, 999999)# 保存真实的短信验证码try:redis_store.setex("sms_code_%s" % mobile, constants.SMS_CODE_REDIS_EXPIRES, sms_code)# 保存发送给这个手机号的记录，防止用户在60s内再次出发发送短信的操作redis_store.setex("send_sms_code_%s" % mobile, constants.SEND_SMS_CODE_INTERVAL, 1)except Exception as e:current_app.logger.error(e)return jsonify(errno=RET.DBERR, errmsg="保存短信验证码异常")# 发送短信# 使用celery异步发送短信, delay函数调用后立即返回（非阻塞）#send_sms.delay(mobile, [sms_code, int(constants.SMS_CODE_REDIS_EXPIRES/60)], 1)# 返回异步任务的对象result_obj = send_sms.delay(mobile, [sms_code, int(constants.SMS_CODE_REDIS_EXPIRES/60)], 1)print(result_obj.id)# 通过异步任务对象的get方法获取异步任务的结果, 默认get方法是阻塞的ret = result_obj.get()print("ret=%s" % ret)# 返回值# 发送成功return jsonify(errno=RET.OK, errmsg="发送成功")

4.client相关代码

# coding:utf-8from celery import Celery
from ihome.libs.yuntongxun.sms import CCP# 定义celery对象
celery_app = Celery("ihome", broker="redis://127.0.0.1:6379/1")@celery_app.task
def send_sms(to, datas, temp_id):"""发送短信的异步任务"""ccp = CCP()ccp.send_template_sms(to, datas, temp_id)# celery开启的命令
# celery -A ihome.tasks.task_sms worker -l info

5.worker相关代码

# coding:utf-8from ihome.tasks.main import celery_app
from ihome.libs.yuntongxun.sms import CCP#
# @celery_app.task
# def send_sms(to, datas, temp_id):
#     """发送短信的异步任务"""
#     pass@celery_app.task
def send_sms(to, datas, temp_id):"""发送短信的异步任务"""ccp = CCP()try:result = ccp.send_template_sms(to, datas, temp_id)except Exception as e:result = -2return result