文章目录

介绍单例类主要作用需求原因实现方法 异步方案生产者消费者设计模式celery介绍和使用介绍实际使用创建环境目录启动异步任务调用异步任务

介绍

本篇博客介绍单例类和生产者消费者两种设计模式两种开发模式,在大多数Python框架中都能用到,我个人的项目主要是django为主,下面讲设计模式中,我可能会略微涉及一点Django,但不会也影响不大。

单例类

主要作用

减少类的实例化次数和内存空间地址的使用,降低内存占用率。

需求原因

首先我们可以看,在没有使用单例类优化的时候,我们如果多次实例化同一个类,这个类会在内存中被创造出两次。如果一个网站的某个功能同时有上百个人在同时使用,可能一些类就会被实例化上百次。但有些类,我们是不需要实例化那么多次的,比如发送短信、邮件验证码的类,生成图像验证码的类,生成词云等,这些功能模块的类很多都只需要实例化一次即可。

实现方法

当前我们的需求是如果我们在实例化一个类的时候,这个类如果被创建了,那我们就可以直接使用,而不是再去实例化一般。 涉及到类的创建,我们需要用到Python类中的一个魔法方法__new__(cls, *args, **kwargs)它是一种负责创建类实例的静态方法,它无需使用 @staticmethod装饰器修饰,且该方法会优先__init__(self)初始化方法被调用。 说人话就是创建类的时候会调用__new__(cls, *args, **kwargs),创建类是在初始化类__init__(self)之前执行的,所以并没有self(代表类的实例化后的自己)而是cls(代表类本身)。 所以我们创建类的时候做一个判断,判断这个类是否已经存在,如果存在就直接返回当前存在的类,而不是再次实例化创建。

class 测试类: def __new__(cls, *args, **kwargs): # hasattr()方法: 返回对象是否具有具有给定名称的属性。 # 判断类属性是否为空对象 if not hasattr(cls, '_实例'): # 如果是空则调用父类方法，为第一个对象创建内存空间 cls._实例 = super().__new__(cls, *args, **kwargs) # 返回类属性保存的对象引用给__init__ return cls._实例 无优化实例化一 = 测试类() 无优化实例化二 = 测试类() print('实例一:', id(无优化实例化一)) print('实例二:', id(无优化实例化二))

注:因为__new__()会优先__init__()初始化方法被调用, 所以使用单例类后,最好就不要再__init__中初始化属性了,如果需要初始化一个值,可以在__new__()中初始化,如果需要像类里传值,就在方法中传入即可。

class 测试类: def __new__(cls, *args, **kwargs): # hasattr()方法: 返回对象是否具有具有给定名称的属性。 # 判断类属性是否为空对象 if not hasattr(cls, '_实例'): # 如果是空则调用父类方法，为第一个对象创建内存空间 cls._实例 = super().__new__(cls, *args, **kwargs) # 返回类属性保存的对象引用给__init__ cls.测试 = 123 return cls._实例 def 测试使用初始化值(self): print(self.测试) def 测试传入值(self, 值): print(值) 单例化一 = 测试类() 单例化一 = 测试类() print('实例一:', id(单例化一)) print('实例二:', id(单例化一)) 单例化一.测试使用初始化值() 单例化一.测试传入值(321)

异步方案

常规情况下,我们的程序是线性的,如果在执行的过程中,有一个结点有较长的延迟,用户就需要再次等待相当长的时间,比如我们通过短信或者邮件的方式给用户发送验证码时,往往会请求第三方服务器进行操作,而这样的操作通常会有秒级延迟。 而发送短信或邮箱验证码的这个任务,我们可以通过解耦,将其从主线程中脱离出来。(除此之外,还有用户文件的格式转换、生成词云等一些可以独立于主线任务但比较耗时的任务进行异步操作。)

生产者消费者设计模式

异步方案中有个比较成熟的设计模式: 生产者消费者设计模式。 此设计模式将异步分为四个模块:任务、生产者、消费者、中间人。 如果想要具体了解这个模式,可以查看这个大佬写的博客,非常形象。

celery介绍和使用

介绍

celery的官方文档

一个简单、灵活且可靠、处理大量消息的分布式系统,可以在一台或者多台机器上运行。单个celery进程每分钟可以处理数以百万计的任务。通过消息进行通信,可以使用消息队列(broker)在生产者和消费者之间进行协调。

简单理解就是:有了celery,我们在使用生产者消费者模式时,只需要关注任务本身,对于线程、队列等的处理优化都交给celery这个库进行即可,极大的简化了程序员的开发流程。

安装

pip install Celery

实际使用

在上述介绍中,我们提到过消息队列,这个消息队列用于存储生产者生成生产资料发送给消费者,所以他需要是一个数据库,这里我采用的是redis,celery当前也支持RabbitMQ, Redis, Amazon SQS等数据库。具体可以去官方文档中查看。

当前假设,我们当前作为一个Django后端,需要给用户发送一个邮件。 任务: 发送邮件 生产者: Django后端 消费者: celery 中间人: 消息队列

创建环境目录

首先,我们在项目目录中创建一个文件夹,用于存放celery的配置和一些任务模块。 其中,需要config.py和main.py作为配置文件 main.py: 这个文件可以原封不动的复制粘贴到你的项目中

# main.py from celery import Celery # celery的入口 # 创建celery实例 celery异步应用 = Celery('celery_app') # 加载配置文件 celery异步应用.config_from_object('celery_tasks.config') # 注册任务 celery异步应用.autodiscover_tasks(['celery_tasks.任务模块']) """ Linux运行 celery -A celery_tasks.main worker -l info • -A指对应的应用程序, 其参数是项目中 Celery实例的位置。 • worker指这里要启动的worker。 • -l指日志等级，比如info等级。 Windows运行 celery -A celery_tasks.main worker -l info --pool=solo """

config.py: 这是这个项目的配置文件,主要是配置数据库的地址,可以根据你是用的数据库进行改写

# config.py # celery配置文件 from Django_Module.settings import Redis数据库 # 指定中间人、消息队列、任务队列 redis # redis的url连接方式-> redis://[:密码]@主机地址(IP地址):端口/数据库号 # 比如 # broker_url = f"redis://:root@127.0.0.1/4" broker_url = f"redis://:{Redis数据库['密码']}@{Redis数据库['地址']}:{Redis数据库['端口']}/4"

任务模块 > tasks.py: 这里是任务模块的启动页面,文件名必须是tasks.py其中写入你需要异步执行的任务,必须如下所示的装饰器装饰。

# 任务模块 > tasks.py # 定义任务 from celery_tasks.任务模块.邮件发送模块.邮件发送库 import 发送邮件 from celery_tasks.main import celery异步应用 # 保证celery识别任务, name是任务的名字 @celery异步应用.task(name="发送异步邮件") def 发送异步邮件(收件邮箱, 发件人='武昌首义学院实验管理平台', 邮件标题='武昌首义学院实验管理平台', 邮件类型=1, 访问链接=None, 自定义内容=None): 返回信息 = 发送邮件(收件邮箱, 发件人, 邮件标题, 邮件类型, 访问链接, 自定义内容) return 返回信息

任务模块 > 邮件发送模块 > 邮件发送库.py: 这是你需要异步执行的文件,这里我只是简单写的一个邮件发送模块。

import random import smtplib from email.mime.text import MIMEText from email.utils import formataddr # 引入全局变量 from Django_Module.settings import 邮箱 # 发送验证模块 def 生成验证码(位数=4): """ 生成大小写字母+数字的随机验证码 :param 位数: 验证码位数(默认=4) :return: 验证码 """ 验证码 = '' # 48-58是ascii的0~9(成五倍是因为数字占比太小,增加数字出现几率),65-90是A~Z,97-122是a~z 随机数ASCII = list(range(48, 58)) * 3 + list(range(65, 91)) + list(range(97, 123)) for i in range(位数): 随机字符 = chr(random.choice(随机数ASCII)) # 65~90为ASCii码表A~Z 验证码 += str(随机字符) return 验证码 def 发送邮件(收件邮箱, 发件人='学院实验管理平台', 邮件标题='学院实验管理平台', 邮件类型=1, 访问链接=None, 自定义内容=None): """ 调用官方邮箱发送邮件 :param 收件邮箱: 接收邮件的邮箱的地址 :param 发件人: 发送人的用户名 :param 邮件标题: 邮件标题 :param 邮件类型: 1:验证邮件,发送六位数验证码。2:邮箱绑定或者找回密码时的操作。3:自定义邮件内容 :param 访问链接: 只有在2时生效, 指定访问的url链接 :param 自定义内容: 只有在3时生效 :return: 发送是否成功(如果邮件类型是1发送成功后会自动返回验证码) """ if 邮件类型 == 1: 验证码 = 生成验证码(6) if 邮件标题 == '学院实验管理平台': 邮件标题 = 邮件标题 + '注册验证码邮件' 邮件内容 = '<p>尊敬的用户,您现在正在进行注册操作,您的验证码为:</p><div style=" margin: auto; text-align:center; font-size: 2rem; color: #333333; background-color: #CCFFFF; width: 20%;">' + 验证码 + '</div><p>验证码五分钟内有效,请尽快使用。</p>' 返回值 = 验证码 elif 邮件类型 == 2: 邮件内容 = f""" <p>尊敬的用户,您现在正在进行邮箱绑定操作,如果确定是您本人操作,请点击:</p> <div style=" margin: auto; text-align:center; font-size: 2rem; color: #333333; background-color: #CCFFFF; width: 20%;"> <a href="{访问链接}" style="font-weight: bold; color: #2c3e50; text-decoration:none;">确认绑定</a> </div> <p>十分钟内有效,超时请重发。</p> <p>如果链接失效请点击: {访问链接}</p> <p>如果此操作非本人进行,请尽快修改密码。</p> """ 返回值 = True elif 邮件类型 == 3: 邮件内容 = 自定义内容 返回值 = None try: 正文 = MIMEText(邮件内容, 'html', 'utf-8') 正文['From'] = formataddr([发件人, 邮箱['使用邮箱']['账号']]) 正文['To'] = 收件邮箱 正文['Subject'] = 邮件标题 server = smtplib.SMTP_SSL(邮箱['使用服务器']['地址'], 邮箱['使用服务器']['端口']) server.login(邮箱['使用邮箱']['账号'], 邮箱['使用邮箱']['密码']) server.sendmail(邮箱['使用邮箱']['账号'], [收件邮箱, ], 正文.as_string()) server.quit() except Exception as e: tag = False else: tag = 返回值 return tag

启动异步任务

Linux运行 celery -A celery_tasks.main worker -l info Windows运行 celery -A celery_tasks.main worker -l info --pool=solo

-A指对应的应用程序, 其参数是项目中 Celery实例的位置。worker指这里要启动的生产者。-l指日志等级，比如info等级。

注: 启动任务的时候确保你所在的路径为任务路径,如果不在可以cd到任务路径,或将celery_tasks.main改为任务的绝对路径

调用异步任务

首先我们需要引入异步任务 from celery_tasks.任务模块.tasks import 发送异步邮件 在需要的地方进行调用(调用的时候一定要加.delay) 正确的调用方式: 发送异步邮件.delay(邮箱, 邮件标题="学院实验管理平台邮箱改绑邮件", 邮件类型=2, 访问链接=链接) 错误的调用方式: 发送异步邮件(邮箱, 邮件标题="学院实验管理平台邮箱改绑邮件", 邮件类型=2, 访问链接=链接)