我使用的框架是flask2.2.0 ，以下操作是基于flask进行

我们先认识一下celery
Celery中文手册
Celery 是一个 基于python的分布式异步任务队列，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery， 举几个实例场景:

异步任务：将耗时的操作任务提交给Celery异步执行，比如发送短信/邮件、消息推送、音频处理等等
做一个定时任务，比如每天定时执行爬虫爬取指定内容
还可以使用celery实现简单的分布式爬虫系统等等
Celery 在执行任务时需要通过一个消息中间件(Broker)来接收和发送任务消息，以及存储任务结果

Celery有以下优点：

简单：Celery 易于使用和维护，并且它 不需要配置文件 ，并且配置和使用还是比较简单的
高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
快速：单个 Celery 进程每分钟可处理数以百万计的任务，而保持往返延迟在亚毫秒级
灵活： Celery 几乎所有部分都可以扩展或单独使用，各个部分可以自定义。

celery核心

1、Task

任务(Task)就是你要做的事情，例如一个注册流程里面有很多任务，给用户发验证邮件就是一个任务，这种耗时任务可以交给Celery去处理，还有一种任务是定时任务，比如每天定时统计网站的注册人数，这个也可以交给Celery周期性的处理。

2、Broker

Broker 的中文意思是经纪人，指为市场上买卖双方提供中介服务的人。在Celery中它介于生产者和消费者之间经纪人，这个角色相当于数据结构中的队列。例如一个Web系统中，生产者是处理核心业务的Web程序，业务中可能会产生一些耗时的任务，比如短信，生产者会将任务发送给 Broker，就是把这个任务暂时放到队列中，等待消费者来处理。消费者是 Worker，是专门用于执行任务的后台服务。Worker 将实时监控队列中是否有新的任务，如果有就拿出来进行处理。Celery 本身不提供队列服务，一般用 Redis 或者 RabbitMQ 来扮演 Broker 的角色

3、Worker

Worker 就是那个一直在后台执行任务的人，也称为任务的消费者，它会实时地监控队列中有没有任务，如果有就立即取出来执行。

4、Beat

Beat 是一个定时任务调度器，它会根据配置定时将任务发送给 Broker，等待 Worker 来消费。

5、Backend

Backend 用于保存任务的执行结果，每个任务都有返回值，比如发送邮件的服务会告诉我们有没有发送成功，这个结果就是存在Backend中，当然我们并不总是要关心任务的执行结果。

邮箱的设置

第一步
先进入邮箱设置（我这里使用的是qq邮箱）

第二步
选择设置里的账户 找到SMTP选项

1
2
3
什么是POP3和SMTP
POP3 用来将邮件下载到本地的一个协议  用于接收邮件
SMTP 是用户来和服务器连接的一个协议  用于发送邮件

这里选择开启 开启后会让用户发送验证码确认用户信息， 之后或返回一个密钥这个密钥复制下来 等会发送邮件的时候要用

需要用到的工具

下载celery

1
pip install celery

安装redis
redis下载连接地址

初始化Celery

在我们的入口文件里设置celery与flask交互

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 这是一个celery的工厂函数，使用flask app中的配置设置celery相关的属性，并且更改了celery对象的Task，
# 使其能够使用flask的应用上下文，这一点非常重要。我们将这段代码放置到flask项目初始化文件中去。
# flask实例和celery实例结合
def make_celery(app):  # 传入flask实例
    # 实例化celery对象    import_name属性
    celery = Celery(app.import_name, broker='redis://localhost:6379/1', backend='redis://localhost:6379/2')
    # 更新配置文件
    celery.config_from_object(config)
    # 类是可以嵌套到方法内部
    # 声明上下文类  celery注入到实例中  把celery和flask结合
    class ContextTask(celery.Task):
        # __call__ 就是一旦被声明 就会被立刻调用
        def __call__(self, *args, **kwargs):
            # 开启上下文
            with app.app_context():
                # 启动任务
                return self.run(*args, **kwargs)
    # 将flask实例和celery的异步任务结合起来
    # 赋值 celery任务= 上下文管理的任务
    celery.Task = ContextTask
    return celery
celery = make_celery(app)

编写发送邮件任务

编写邮箱任务类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
#异步任务库
#导入入口文件实例化好的celery
from manage.py import celery
# 在发送邮件的过程中起到服务器之间互相通信的作用。
import smtplib  
#导入文件文本  用来构建内容   写邮件、添加附件是由 email 模块控制。
from email.mime.text import MIMEText  
#导入邮件分类  用来设置邮件体
from email.mime.multipart import MIMEMultipart

@celery.task() #装饰器 用来创建任务
#发送邮箱验证码操作
class SendEmail:

    #初始化方法  初始化的时候进行赋值操作
    def __init__(self):
        #初始化邮箱数据
        #发送人的邮箱  也就是你自己的邮箱
        self._user = '*****'
        #发送密钥  也就是授权码
        self._pwd = '*****'

    #发送方法  self关键字  用来获取初始化变量
    #_touser 收件人  _title 邮件标题   _content  邮件内容
    def send_mail(self,_touser,_title,_content):
        print('发送邮件')
        #构建邮件体
        msg = MIMEMultipart()
        #邮件标题
        msg['Subject'] = _title
        #发件人
        msg['From'] = self._user
        #收件人
        msg['To'] = _touser
        #构建内容
        # MIMEText(msg,type,chartset)
        # msg：文本内容，可自定义
        # type：文本类型，默认为plain（纯文本）
        # chartset：文本编码，中文为“utf-8”
        part = MIMEText(_content,'html','utf-8')
        #attach 将对象和资源句柄联系起来
        msg.attach(part)

        #发送逻辑
        #建立连接对象
        s = smtplib.SMTP_SSL('smtp.qq.com',465)
        #登录邮箱  用来登录
        s.login(self._user,self._pwd)
        #发送操作   as_string是把MIMEText对象变成str
        s.sendmail(self._user,_touser,msg.as_string())
        #关闭连接
        s.close()

使用异步任务

在我们的视图中使用异步任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#导入异步任务库
from task import SendEmail
# decode_responses  将接收到的数据自动解码
r = redis.Redis(decode_responses=True)
#生成随机验证码
import random
import string
#正则判断邮箱是否正确
import re

class SendCodeApi(MethodView):
    #发送邮箱验证码
    #导入异步任务库
    def get(self):
        #接收邮箱参数
        email = request.args.get('email',None)
        
        # r.exists是判断该邮箱是否存在redis 
        #如果该邮箱验证码已经存在redis  那就代表你上一次发送验证码成功 必须在验证码过期后才能继续发送
        code = r.exists(‘black_’+email)
        if code:
        #ttl 获取key的过期时间
            times = r.ttl(email)
            return jsonify({
                'code':204,
                'msg':'请在%s秒后重试'%times,
                'time':times
            })

        if not email:
            return jsonify({
                'code':203,
                'msg':'请输入邮箱'
            })

        if not re.fendall(r'^[1-9a-z][0-9a-z]{4,15}@qq\.com$',emial):
            return jsonify({
                'code':204,
                'msg':'请输入合法的邮箱'
            })
            
        #获取计数器
        #该操作是为了限制用户发送邮箱验证码次数，用户一旦在24小时内发送邮箱验证码超过三次 将会把该用户加入黑名单24小时
        sendcode = r.get(email+'_count')  #获取邮箱的发送次数
        if sendcode and int(sendcode)>3:
            return jsonify({
                'code':204,
                'msg':'已达到邮箱发送次数的最大上限'
            })

        #生成随机验证码  随机获取4位的整数或者小数类型数据 通过join拼接为字符串
        code = course =''.join(random.sample(string.digits + string.ascii_lowercase,4))

        #发送邮件
        try:
            #异步发送邮箱验证码 调用异步任务库
            #delay是调用异步任务的方法  后面是发送邮箱需要的参数
            taskss.SendEmail.delay(email, '社交平台验证码', code)
        except Exception as e:
            return jsonify({
                'code':205,
                'msg':'邮件发送失败'
            })
        #保存验证码
        #将验证码存入redis进行保存  ex设置过期时间为500秒
        r.set(email,code,ex=500)

		#获取计数器操作 
        if not r.get(email+'_count'):
             # 获取不到就代表是第一次发送验证码 值为一
             # 就将值存入redis 
              r.set(email+'_count',1)
         else:
             # 这次获取到了 代表第二次了  要进行累加  等第三次还是获取的到  再次进行累加
             #incr redis的累加值操作
             r.incr(email+'_count')
             
         return jsonify({
             'code': 200,
             'msg': '邮箱验证码发送成功'
         })

如何启动异步

第一步先启动redis

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
#进入redis根目录，输入下面的指令启动服务器：
redis-server.exe redis.windows.conf

C:\renaissace\redis\Redis-x64-3.2.100>redis-server.exe redis.windows.conf
                _._
           _.-``__ ''-._
      _.-``    `.  `_.  ''-._           Redis 3.2.100 (00000000/0) 64 bit
  .-`` .-```.  ```\/    _.,_ ''-._
 (    '      ,       .-`  | `,    )     Running in standalone mode
 |`-._`-...-` __...-.``-._|'` _.-'|     Port: 6379
 |    `-._   `._    /     _.-'    |     PID: 2804
  `-._    `-._  `-./  _.-'    _.-'
 |`-._`-._    `-.__.-'    _.-'_.-'|
 |    `-._`-._        _.-'_.-'    |           https://redis.io
  `-._    `-._`-.__.-'_.-'    _.-'
 |`-._`-._    `-.__.-'    _.-'_.-'|
 |    `-._`-._        _.-'_.-'    |
  `-._    `-._`-.__.-'_.-'    _.-'
      `-._    `-.__.-'    _.-'
          `-._        _.-'
              `-.__.-'

[2804] 13 Sep 11:59:02.735 # Server started, Redis version 3.2.100
[2804] 13 Sep 11:59:02.736 * DB loaded from disk: 0.000 seconds
[2804] 13 Sep 11:59:02.739 * The server is now ready to accept connections on po
rt 6379
显示以上内容既代表redis启动成功

第二步启动celery

1
2
#taskss是我的异步任务库文件名
celery -A taskss.celery worker --loglevel=info

以上就是异步发送验证码的全部过程