django-celery异步发送邮箱

celery介绍

将redis发布订阅模式用做消息队列和rabbitmq的区别：

可靠性

redis：没有相应的机制保证消息的可靠消费，如果发布者发布一条消息，而没有对应的订阅者的话，这条消息将丢失， 不会存在内存中；rabbbitmq： 具有消息消费确认机制，如果发布一条消息，还没有消费者消费该队列，那么这条消息将一直存放在队列中，直到有消费者消费了该条消息，以此可以保证消息的可靠消费。

实时性

redis实时性高，redis作为高效的缓存服务器，所有数据都存在在服务器中，所以它具有更高的实时性消费者负载均衡；rabbitmq队列可以被多个消费者同时监控消费，但是每一条消息只能被消费一次，由于rabbitmq的消费确认机制，因此它能够根据消费者的消费能力而调整它的负载，redis发布订阅模式，一个队列 可以被多个消费者同时订阅，当有消息到达时，会将该消息依次发送给每个订阅者；

持久性

redis：redis的持久化是针对 于整个redis缓存的内容，它有RDB和AOF两个持久化方式，可以将整个redis实例持久化到磁盘，以此来做数据备份，防止异常情况下导致数据丢失。rabbitmq：队列、消息都可以选择性持久化，持久化粒度更小，更灵活；队列监控rabbitmq实现了后台监控平台，可以在该平台上看到所有创建的队列的详细情况，良好的台后管理平台可以方便我们更好的使用；redis没有所谓的监控平台。

总结：

redis：轻量级、低延迟，高并发、低可靠性；

rabbitmq：重量级、高可靠，异步，不保证实时

rabbitmq是一个专门的AMQP协议队列， 它的优势就在于提供可靠的队列服务，并且可做到异步，而redis主要是用于缓存的，redis的发布订阅模块，可用于实现及时性，且可靠性低的功能

名词

task(任务)： 就是一个Python函数

queue(队列)：将需要执行的任务加入到队列中

worker(工人)：在一个新进程中，负责执行队列中的任务

borker(代理人)：负责调度，在布置环境中使用redis

使用

安装包

• pip install celery==3.1.25
• pip install celery-with-redis==3.0
• pip install django-celery==3.3.0
• pip install django-redis==4.10.0
• pip install redis==2.10.6
• pip install itsdangerous == 1.1.0

在settings中配置

# INSTALLED_APPS中安装应用
ISNTALLED_APPS = [
    'djcelery',
]

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

在项目下，创建celery_tasks文件夹，在里面创建文件，init.py, celery.py，celeryconfig.py以及tasks.py文件

celery.py下写入的内容

from celery import Celery
from celery_tasks import celeryconfig # 导入celery配置文件

import os
# 为celery设置环境变量
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mydemo.settings")

## 创建celery app
app = Celery('celery_tasks')

# 从单独的配置模块中加载配置
app.config_from_object(celeryconfig)

# 设置app自动加载任务
app.autodiscover_tasks(['celery_tasks'])

celeryconfig.py内写入的内容

# 设置结果存储
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/9'

# 设置代理人broker
BROKER_URL = 'redis://127.0.0.1:6379/8'

tasks.py写入的内容

from celery_tasks.celery import app as celery_app  # 导入创建好的celery应用
from django.core.mail import send_mail  # 使用django内置函数发送邮件
from django.conf import settings  # 导入django的配置


@celery_app.task
def send_mail_task(email, token):
    # 使用django内置函数发送邮件
    subject = "海上明月"
    message = ""
    sender = settings.EMAIL_FROM
    recipient = [email]
    html_message = "<h1>{}恭喜您成为美多商城会员，请点击以下链接进行激活邮箱：<a href='http://127.0.0.1:8000/mdadmin/active/{}'>
    http://127.0.0.1:8000/mdadmin/active/{}</a></h1>".format("海上明月", token, token)
    send_mail(subject, message, sender, recipient, html_message=html_message)

views.py

from django.http import HttpResponse
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from itsdangerous import SignatureExpired
from celery_tasks.tasks import send_mail_task
class SendEmailView(View):
    """
    发送邮件
    """
    def post(self, request):
        print(request.POST)
        email = request.POST.get('email')
        user_info = {"user": 1}
        serialize = Serializer(settings.SECRET_KEY, 1800)
        token = serialize.dumps(user_info).decode()
        send_mail_task.delay(email, token)
        return HttpResponse(json.dumps({'msg': 'OK', 'code': 200}))

激活用户

from itsdangerous import SignatureExpired


class ActiveView(View):
    """
    激活用户
    """
    def get(self, request, token):
        serialize = Serializer(settings.SECRET_KEY, 60)
        try:
            user_info = serialize.loads(token)
            print(user_info)
            user_id = user_info["user"]
            print(user_id)
            user = User.objects.get(pk=user_id)
            user.is_active = True
            user.save()
            return HttpResponse(json.dumps({'msg': '激活成功', 'code': 200}, ensure_ascii=False))
        except SignatureExpired:
            return HttpResponse(json.dumps({'msg': '激活链接失效，请重新发送邮件进行激活', 'code': 404}, ensure_ascii=False))

启动celery服务进行测试

# 启动celery服务进行测试
# 启动django服务
python manage.py runserver
# 启动celery的worker
celery -A celery_tasks worker -l info -P eventlet