异步底层代码实现邮件发送

1 栈和队列

1.1 队列

class MyQueue:
    # 默认继承object类，python3里面不需要写
    def __init__(self):
        # def __init__(self, s):带参，初始化需要传参
		# 是初始化方法，没有返回值，__new__是构造方法，有返回值。构造方法优先于初始化列表
       	self.s = []
    def push(self, x:int) -> None:
        # 声明类型，减少bug，提高代码可读性
        # 箭头方法：代表返回值，什么也不返回，方法被建立后，返回什么
        self.s.append(x)
    def pop(self) -> int:
        # 拿到值做任务（task_id————>任务id）
        return self.s.pop(0)
    def empty(self) -> bool:
        return not bool(self.s)

myq = MyQueue()
my.push(1)
print(myq.pop())
print(myq.empty())

1.2 栈

class MyStack:
    def __init__(self):
		self.s = []
    def push(self, x:int) -> None:
        self.s.append(x)
    def pop(self) -> int:
        return sekf.s.pop()
    	# 默认从最后删除
    def size(self) -> int:
        return len(self.s)
    def empty(self) -> bool:
        return self.s == []
    
# 位运算，十进制转换二进制

def transform(num:int) -> str:
    stack1 = MyStack()
    while num != 0:
        remain = num % 2			# 取余
        num = int(num/2)
        stack1.push(remain)			# 入栈
    w = ''
    while not stack.empty():
        s += str(stack1.pop())
    return s
print(transform(13))

2 异步引入

2.1 导入原因

由于django自身模型导致等待任务，阻塞，所以引入异步这个概念。针对于celery模型的思考，其实质就是一个消息队列，采用的是异步消费。异步任务队列是一个容器，遵循的是先进先出，继承的是基础类。针对于异步操作，我们可以引入多进程，多线程以及协程实现。但是针对于性能的考虑，我首先引入了多线程，但是线程切换仍然十分浪费资源，最合适的应该是协程。

2.2 思考导入

import datetime
import schedule
import threading
import time


def job1():
    print("I'm working for job1")
    time.sleep(2)
    print("job1:", datetime.datetime.now())


def job2():
    print("I'm working for job2")
    time.sleep(2)
    print("job2:", datetime.datetime.now())


def job1_task():
    threading.Thread(target=job1).start()


def job2_task():
    threading.Thread(target=job2).start()


def run():
    schedule.every(1).seconds.do(job1_task)
    schedule.every(1).seconds.do(job2_task)

    while True:
        schedule.run_pending()
        time.sleep(1)
run()

2.3 静态方法

• 使用的注意事项

class Action:
    def __init__(self):
        self.backend = [1, 2, 3, 4, 5]
    # @staticmethod         # 使用 @staticmethod方法，不再使用类中的属性
    def act(self):
        start = time.time()
        for i in self.backend:
            print(i)
            time.sleep(1)
        end = time.time()
        return ('运行秒数：', end-start)
    @staticmethod
    def tell():
        print('123')

a = Action()
print(a.act())
Action.tell()		# 不需要实例化对象和方法，直接调用

2.4 底层代码实现异步

2.4.1 多线程（第一次思考）

• 代码

# asynchronous

# backend消息队列--------------------------------------------

class BackendQueue:
    def __init__(self):
        self.queue_list = []
    def push(self, x:int) -> None:
        self.queue_list.append(x)
    def pop(self) -> int:
        return self.queue_list.pop(0)
    def empty(self) -> bool:
        return not bool(self.queue_list)

# myq = BackendQueue()
# myq.push(1)         # 存入任务id
# myq.push(2)
# myq.push(3)
# myq.push(4)
# # myq.queue_list是backend中的数据
# task_list = []            # 获取到的任务id列表
# for i in myq.queue_list:    # 从任务id列表中取出任务id交付 worker
#     task_list.append(i)


# worker执行任务-----------------------------------------------------
def asynchronous_task(task):
    print('任务id:', task, '正在执行')

# 采用多线程，worker异步执行
import threading
def asynchronous_worker(task_list):
    for i in range(len(task_list)):     # 根据列表长度设置相应线程数
        task = task_list[i]             # 获取任务id
        t = threading.Thread(target=asynchronous_task(task))
        t.start()
        print('任务执行结果：', task, '执行完毕')              # 执行结果存入redis


# 调函数-------------------------------------------------------

def runQueue():
    myq = BackendQueue()
    myq.push(1)  # 存入任务id
    myq.push(2)
    myq.push(3)
    myq.push(4)
    # myq.queue_list是backend中的数据
    task_list = []  # 获取到的任务id列表
    for i in myq.queue_list:  # 从任务id列表中取出任务id交付 worker
        task_list.append(i)
    asynchronous_worker(task_list)

runQueue()

• 结果

redis:-->[1, 2, 3, 4]
任务id: 1 正在执行
任务执行结果： 1 执行完毕
任务id: 2 正在执行
任务执行结果： 2 执行完毕
任务id: 3 正在执行
任务执行结果： 3 执行完毕
任务id: 4 正在执行
任务执行结果： 4 执行完毕
redis:-->{
    '1':'成功',
    '2':'成功',
    '3':'成功',
    '4':'成功',
         }

2.4.2 多线程（课上代码优化）

import redis

class RedisQueue:
    def __init__(self, key, **redis_kwargs):
        # **redis_kwargs 为字典，不定长参数，可传可不传（如果是docker，一定要传，因为docker是端口映射技术）
        # key是形参，必须传
        # *args必须死元祖或者列表
        self.__db = redis.Redis(**redis_kwargs)
        # __db私有变量
        self.key = key
    def qsize(self):
        return self.__db.llen(self.key)
    	# 返回队列长度
        # list适合任务队列 
    def put(self, item):
        self._db.rpush(self.key, item)
        # rpush 代表从右侧入队
        # lpush 代表从左边入队
    def pop(self):
        item = self.__db.lpop(self.key)
        return item

# q = RedisQueue('mykey')
# for i in range(5):
#     q.push()		# 消费换成 pop()，异步消费
#     print(i)
#     time.sleep(1)
# print(q.qsize())		# celery底层，没有执行，就放入

import threading
def dojob():
    q = ReidsQueue('mykey')
    while 1:
        result = q.pop()
        if not result:
            break
        time.sleep(1)
        
for index in range(2):
    thread = threading.Tread(target=dojob)
    thread.start()

3 Django中异步实现

3.1 我的思考

• 线程切换十分耗时，所以我们采用协程，更加轻量级
• 协程是并发（多进程多线程是并行）
• 经过深思熟虑，后来发现，不用传递参数，只是执行任务的时候，所需参数在数据库中存储的id不同，所以只需要绑定taskid和用户id放入redis，就可以实现这个效果啦！！！

3.2 实现代码

3.2.1 send_email.py

# 邮箱配置
# 配置邮件发送
from django.conf import settings
from django.core.mail import send_mail

from userapp.models import UserModel



def EmailInform(user_id):
        email = UserModel.objects.filter(pk=user_id).first().email
        print(123, email)
        username = UserModel.objects.filter(pk=user_id).first().username
        print(234, username)
        subject = '一个可爱的通知(^o^)/~'
        message = ''
        from_email = settings.EMAIL_FROM
        recipient_list = [email]
        html_message = 'dear{},您有新的通知信息了！'.format(username)
        send_mail(subject=subject,
                  message=message,
                  from_email=from_email,
                  recipient_list=recipient_list,
                  html_message=html_message)

3.2.2 asynchronous.py

# 协程----------------------------------------------------------------------------------

# 把任务放入消息队列，本次定义2个任务一执行
# 存入任务id的队列是 5号库
# 存入任务结果的队列是 6号库



import redis
from django.http import HttpResponse

from noticeapp.send_email import EmailInform


class NoticeQueue():
    def __init__(self, key):
        self.__db = redis.Redis(db=5, decode_responses=True)
        self.key = 'queue'
    def push(self, item):
        # 从右侧入队
        self.__db.rpush(self.key, item)
    def pop(self):
        # 从左侧出队
        return self.__db.lpop(self.key)
    def qsize(self):
        # 查看队列长度
        return self.__db.llen(self.key)

# 执行任务的worker



import gevent
def email_worker(request):
    # 存放任务结果
    r = redis.Redis(db=6, decode_responses=True)
    key = request.GET.get('user_id')
    # 也可以写获取用户token，通过解码再获取
    q = NoticeQueue('queue')
    q.push(key)
    while True:
        if q.qsize() == 0:      # 没有任务直接退出
            break
        elif q.qsize() >= 2:    # 每两个执行一次，执行成功，就在列表中删除任务id
            user_id1 = int(q.pop())
            user_id2 = int(q.pop())
            print(user_id1)
            print(user_id2)
            t1 = gevent.spawn(EmailInform, user_id1)
            t2 = gevent.spawn(EmailInform, user_id2)
            t1.join()       # 等待协程执行完毕
            t2.join()
            r.set(user_id1, '成功')
            r.set(user_id2, '成功')
            return HttpResponse('okok')
        else:
            user_id = int(q.pop())
            t = gevent.spawn(EmailInform, user_id)
            t.join()
            r.set(user_id,'成功')
        return HttpResponse('等待执行')

3.2.3 urls.py

from django.urls import path

from noticeapp.asynchronous import email_worker

urlpatterns = [
    path('email_worker/', email_worker)
]

• 还是有一点小小的bug，但是基础都已经实现了，同时接到很多的时候，就可以异步啦，测试的时候注释掉一个的效果！