- 环境
requests==2.32.3
Django==2.1.15
celery==5.4.0
redis==3.4.1
django-redis==4.11.0
djangorestframework==3.11.0
djangorestframework-jwt==1.11.0
eventlet==0.36.1
注:使用redis作为缓存,restframework框架,jwt token
2.celery异步配置
1.在项目的根目录新建文件夹 celery_tasks
2.在celery_tasks下新建config.py 配置文件
# config.py
# 编写celery配置信息
# 指定消息队列的位置
broker_url = 'redis://127.0.0.1:6379/15'
# 返回执行结果
result_backend = 'redis://127.0.0.1:6379/14'
3.在celery_tasks下新建main.py,celery启动文件
# main.py
from celery import Celery
import os
# 创建celery实例对象
celery_app = Celery('frame')
# 把celery和django组合,识别加载django 的配置
os.environ["DJANGO_SETTINGS_MODULE"] = "Frame.settings"
# 通过app对象加载配置
celery_app.config_from_object('celery_tasks.config')
# 加载任务
# 参数必须是一个列表
celery_app.autodiscover_tasks(['celery_tasks.sms', ])
4.在celery_tasks下新建sms/tasks.py文件,添加任务
# tasks.py
from celery_tasks.main import celery_app
from celery_tasks.sms.menwang.smsclient import SmsClient
import logging
logger = logging.getLogger("django")
@celery_app.task(name='send_sms_code')
def send_sms_code(mobile, sms_code):
try:
send_res = SmsClient().singleSend(mobile=mobile, code=sms_code, expiration_time=5)
# 调用云通讯接口发送短信,接受手机号,验证码,过期时间
except Exception as e:
raise Exception(e)
return send_res
项目目录结构
3.梦网云通讯短信服务。免费赠送100条短信,直接在官网注册登录,个人测试可以不用资质认证
1.在文本短信---发送账号页下拉,点击“立刻获得测试账号”,系统即生成一个测试账号,在列表中展示。
url、用户名、密码、apikey获取
短信模板获取
准备工作完毕,接下来连接云通讯发送短信
4.在sms文件夹下新建menwang/smsclient.py发送单条短信
import time
import json
import traceback
import requests
from celery_tasks.sms.menwang import smsmessage
from celery_tasks.sms.menwang.smsexception import *
# 文本短信发送客户端
class SmsClient():
def __init__(self):
self._userid = 'RTFG56' # 发送者帐号
self._pwd = 'UYIJK89' # 发送者帐号的密码
self._url = 'http://api01.monyun.cn:7911/sms/v2/sdd/' # 请前往您的控制台获取请求域名(IP)或联系梦网客服进行获取
@property
def userid(self):
return self._userid
@property
def pwd(self):
return self._pwd
@property
def url(self):
return self._url
# http post
def postSmsMessage(self, message):
fullurl = self.url + message.apiname
try:
r = None
body = message.toJson()
timeout = (5, 30)
headers = {'Content-Type': 'application/json', 'Connection': 'Close'}
# 短连接请求
r = requests.post(fullurl, data=body, headers=headers, timeout=timeout)
r.encoding = 'utf-8'
debugStr = '\n[------------------------------------------------------------\n' + \
'http url:' + fullurl + '\n' + \
'headers:' + headers.__str__() + '\n' + \
body + '\n' + \
'status code:' + str(r.status_code) + '\n' + \
r.text + \
'\n-------------------------------------------------------------]\n'
print(debugStr)
# http请求失败
if (r.status_code != requests.codes.ok):
return message.makeupRet(SmsErrorCode.ERROR_310099)
# 请求成功,解析服务器返回的json数据,
rTest = json.loads(r.text)
return rTest
except SmsValueError as v:
return message.makeupRet(v.errorcode)
except requests.RequestException as e:
print(traceback.format_exc().__str__())
return message.makeupRet(SmsErrorCode.ERROR_310099)
except Exception as e:
print(traceback.format_exc().__str__())
return message.makeupRet(SmsErrorCode.ERROR_310099)
# 单条发送(短信)
def singleSend(self, mobile, code, expiration_time):
message = smsmessage.SmsSingleMessage()
# 发送者帐号
message.userid = self.userid
# 密码
message.pwd = self.pwd
# 接收方手机号码
message.mobile = mobile
# 验证码数字<=6位
message.content = u'您的验证码是{0},在{1}分钟内输入有效。如非本人操作请忽略此短信。'.format(code, expiration_time)
# 业务类型:最大可支持10个长度的ASCII字符串:字母,数字
message.svrtype = '0123456789'
# 扩展号:长度不能超过6位,注意通道号+扩展号的总长度不能超过20位,若超出exno无效,如不需要扩展号则不用提交此字段或填空
message.exno = '123456'
# 用户自定义流水编号:该条短信在您业务系统内的ID,比如订单号或者短信发送记录的流水号。填写后发送状态返回值内将包含这个ID。
# 最大可支持64位的ASCII字符串:字母、数字、下划线、减号,如不需要则不用提交此字段或填空
message.custid = 'b3d0a2783d31b21b8573'
# 业务类型:最大可支持10个长度的ASCII字符串:字母,数字
message.exdata = '0123456789'
ret = self.postSmsMessage(message=message)
print('singleSend:', ret)
return ret
别的接口可以到官网下载demo,里面有各种详细的示例
5.启动celery
# 在项目的根目录下
celery -A celery_tasks.main worker --loglevel=info
# 以守护进程方式启动
celery multi start work -A main -l info -B --logfile=celerylog.log
注:在Windows系统中发现任务启动了,redis中也有记录,但就是不发短信。原因:celery不支持在windows下运行任务,需要借助eventlet来完成
前面已经安装过了
celery -A celery_tasks.main worker -l info -P eventlet -c 10
6.最后创建appp,挂载、路由、这些就不说了,在视图views中调用celery任务异步发送短信
from django_redis import get_redis_connection
from rest_framework.views import APIView, Response
from random import randint
from celery_tasks.sms.tasks import send_sms_code
# Create your views here.
ret = {'code': 100, 'msg': None, 'data': None}
class SMSCode(APIView):
authentication_classes = [] # 不进行token认证
def post(self, request):
global ret
try:
mobile = request.data['mobile']
redis_conn = get_redis_connection('verify_codes')
send_flag = redis_conn.get('send_flag_%s' % mobile)
if send_flag:
ret['code'] = 101
ret['msg'] = '请勿频繁发送短信'
return Response(ret)
sms_code = '%06d' % randint(0, 999999)
result = send_sms_code.delay(mobile=str(mobile), sms_code=sms_code)
print(result) # 任务处理结果task_id
# SmsClient().singleSend(str(mobile), sms_code, SMS_CODE_REDIS_EXPIRES // 60)
pl = redis_conn.pipeline()
pl.setex('sms_code_%s' % mobile, 300, sms_code)
pl.setex('send_flag_%s' % mobile, 60, 1)
pl.execute()
ret['msg'] = '短信发送成功'
except Exception as e:
ret['code'] = 102
ret['msg'] = e
return Response(ret)