完成celery简单发送注册邮件

这个是基于Django原始的celery

启动celery指令

celery -A celery_test worker -l info -P eventlet

大部分代码,具体代码看业务

在django项目celery_test/celery_test/下创建celery.py文件，配置以下内容：

import os
from celery import Celery
# 把celery和django进行组合，识别和加载django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'opwf.settings')
# 创建celery实例
app = Celery('opwf')
# 指定celery消息队列的配置
app.config_from_object('opwf.config', namespace='CELERY')
# 从所有的django-app中加载任务
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

在django项目celery_test/celery_test/下创建config.py文件，配置以下内容：

# 消息中间人设置
broker_url = 'redis://127.0.0.1:6379/15'
# 结果存储设置
result_backend = 'redis://127.0.0.1:6379/14'

在django项目celery_test/celery_test/下 init.py 中写入以下内容：

# 绝对引用，使我们的celery模块不会与原始的celery冲突
from __future__ import absolute_import, unicode_literals
# 加入绝对引入以后，导入当前模块下的内容方法： from xx import xx as xx
from .celery import app as celery_app

__all__ = ('celery_app',)

在settings.py中配置发邮件的内容：

# 配置邮件发送
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.163.com'  # 如果为163邮箱，设置为smtp.163.com
EMAIL_PORT = 25  # 或者 465/587是设置了 SSL 加密方式
# 发送邮件的邮箱
EMAIL_HOST_USER = 'wangsai_python@163.com'
# 在邮箱中设置的客户端授权密码
EMAIL_HOST_PASSWORD = 'LSJSTWQLCQMAWEWX'  # 第三方登陆使用的授权密码
EMAIL_USE_TLS = True  # 这里必须是 True，否则发送不成功
# 收件人看到的发件人, 必须是一直且有效的
EMAIL_FROM = '公司老董<wangsai_python@163.com>'
DEFAULT_FROM_EMAIL = EMAIL_HOST_USER

在django的app中创建tasks.py文件，在这里写入异步任务内容：

# -*- coding: utf-8 -*-
# 绝对引用，使我们的celery模块不会与原始的celery冲突
from __future__ import absolute_import, unicode_literals
# 导入原始的celery模块中shared_task from xx import xx
from celery import shared_task
# 使用django内置函数发送邮件
from django.core.mail import send_mail
# 导入django的settings
from django.conf import settings


@shared_task
def send_mail_task(username,email):
    subject = '工单通知'
    message = ''
    from_email = settings.EMAIL_FROM
    recipient_list = [email]
    html_message = '<h1>工单审批通知</h1>'.format(
        username,)
    send_mail(subject, message, from_email, recipient_list, html_message=html_message)

在django的app下的views.py中调用tasts下的任务

token_serializer = TimedJSONWebSignatureSerializer(settings.SECRET_KEY, 1800)
login_serializer = TimedJSONWebSignatureSerializer(settings.SECRET_KEY, 86400)
from myutils.tasks import send_meail_task
from myutils.tasks import send_meail_task
from django.conf import settings

class UserView(APIView):
    def post(self, request):
        username = request.data.get('username')
        password: object = request.data.get('password')
        pwd = request.data.get('pwd')
        phone = request.data.get('phone')
        email = request.data.get('email')
        rola = request.data.get('rola')

        # 对手机号进行验证
        phone_re = re.compile(r'^1(3d|4[4-9]|5[0-35-9]|6[67]|7[013-8]|8[0-9]|9[0-9])d{8}$')
        if not re.search(phone_re, phone):
            return Response({'msg': '手机号不符合规范', 'code': 400})
        # 对用户名进行验证
        users = User.objects.filter(username=username)
        if users:
            return Response({'code': 300, 'msg': '用户名重复'})
        # 对密码进行验证
        if password != pwd:
            return Response({'code': 300, 'msg': '第一次密码与第二次密码错误'})

        # 构造数据
        data = {
            'username': username,
            'password': make_password(password),
            'pwd': pwd,
            'phone': phone,
            'email': email,
            'rola': rola,
        }
        # 对数据进行序列化
        obj_all = UserSer(data=data)
        # 对数据进行验证
        if obj_all.is_valid():
            obj_all.save()
            # 同步发邮箱
            user_info = {'user_id': obj_all.data.get('user_id')}
            token = token_serializer.dumps(user_info).decode()
            send_meail_task.delay(username, token, email)
            r.incr(username)
            return Response({'code': 200, 'msg': '注册成功'})
        else:
            print(obj_all.errors)
            return Response({'code': 400, 'msg': '注册失败'})

需要的model类