web项目邮件的发送必不可少，这里推荐两种在flask下的邮件的异步发送。
1：使用thread模块：
优点：配置方便，代码简单。
缺点：flask项目部署在uswgi下的时候，因为pythongil的原因会失效，造成邮件无法发送。
email.py，主要用于定义邮件
#文件:email.py
from flask_mail import Message, Mail
from flask import render_template
from exts import mail
from flask import current_app
from threading import Thread
#to 接收方
#sender 发送方
#subject 主题
#template 内容
def send_async_mail(to,subject,template,**kwargs):#发送邮件
msg = Message(current_app.config["FLASKY_MAIL_SUBJECT_PREFIX"] + subject,sender = current_app.config['FLASKY_MAIL_SENDER'],recipients=[to])
# msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
mail.send(msg)
def send_async_email(app,msg):
with app.app_context():
for i in range(0,5):
try:
mail.send(msg)
return True
except Exception as e:
i =+ 1
print("err is {} .index is {}\n".format(e,i))
return False
def send_email(to,subject,template,**kwargs):#发送邮件
try:
msg = Message(current_app.config["FLASKY_MAIL_SUBJECT_PREFIX"] + subject,sender = current_app.config['FLASKY_MAIL_SENDER'],recipients=[to])
msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
app = current_app._get_current_object() #这一句非常重要，因为flask的上下文特性，如果没有这一句会造成邮件无法发送
thr = Thread(target=send_async_email,args= [app,msg])
thr.start()
return thr
except Exception as e:
print(e)
2：使用celery：
优点：flask项目部署在uswgi下的时候，邮件仍可发送，代码简单。。
缺点：配置相对复杂，
一.python层面工作
tasks.py，主要用于定义邮件
#tasks.py
from celery import Celery
from flask_mail import Message,Mail
from flask import Flask,render_template
import config
mail = Mail()
app = Flask(__name__)
app.config.from_object(config)
mail.init_app(app)
# 运行本文件：
# 在windows操作系统上：
# celery -A tasks.celery worker --pool=solo --loglevel=info
# 在类*nix操作系统上：
# celery -A tasks.celery worker --loglevel=info
def make_celery(app):
celery = Celery(app.import_name, backend=app.config['CELERY_RESULT_BACKEND'],
broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
TaskBase = celery.Task
class ContextTask(TaskBase):
abstract = True
def __call__(self, *args, **kwargs):
with app.app_context():
return TaskBase.__call__(self, *args, **kwargs)
celery.Task = ContextTask
return celery
celery = make_celery(app)
@celery.task
def send_email(to,subject,template,**kwargs):#发送邮件
msg = Message(app.config["FLASKY_MAIL_SUBJECT_PREFIX"] + subject,sender = app.config['FLASKY_MAIL_SENDER'],recipients=[to])
msg.html = render_template(template + '.html', **kwargs)
mail.send(msg)
view层的调用
#view.py
......
from tasks import send_email
......
send_email.delay(email,"请查收您的验证码","auth/email/confirm",email = email,captcha = captcha)
......
二.服务器配置
安装redis
sudo apt-get install redis-server #安装
sudo apt-get purge --auto-remove redis-server #完全卸载
ps aux|grep redis #启动：redis安装后，默认会自动启动，可以通过以下命令查看：
sudo service redis-server start # 如果想自己手动启动，可以通过以下命令进行启动：
sudo service redis-server stop # 停止
安装celery
pip install celery #安装
celery -A tasks.celery worker --loglevel=info #启动
