单例模式
1.介绍
单例模式(Singleton Pattern) 是 java 中最简单的设计模式之一。这种类型的设计模式属于创建型模式,它提供了一种创建对象的最佳方式。
单例模式涉及到一个单一的类,该类负责创建自己的对象,同时确保只有单个对象被创建。这个类提供了一种访问其唯一的对象的方式,可以直接访问,不需
要实例化该类的对象。
注意:
1. 单例类只能有一个实例。
2. 单例类必须自己创建自己的唯一实例
3. 单例类必须给所有的其它对象提供这一实例。
意图: 保证一个类仅有一个实例,并提供一个访问它的全局访问点。
主要解决:一个全局使用的类频繁地创建与销毁。
何时使用:当您想要控制实例数目,节省系统资源的时候。
如何解决: 判断系统是否已经有这个实例,如果有就返回,如果没有就创建。
应用实例:
1.一个党只能有一个主席
2、Windows 是多进程多线程的,在操作一个文件的时候,就不可避免地出现多个进程或线程同时操作一个文件的现象,所以所有文件的处理必须通过
唯一的实例来进行。
3、一些设备管理器常常设计为单例模式,比如一个电脑有两台打印机,在输出的时候就要处理不能两台打印机打印同一个文件。
优点:
1、在内存里只有一个实例,减少了内存的开销,尤其是频繁的创建和销毁实例(比如管理学院首页页面缓存)。
2、避免对资源的多重占用(比如写文件操作)。
缺点:没有接口,不能继承,与单一职责原则冲突,一个类应该只关心内部逻辑,而不关心外面怎么样来实例化。
使用场景:
1、要求生产唯一序列号。
2、WEB 中的计数器,不用每次刷新都在数据库里加一次,用单例先缓存起来。
3、创建的一个对象需要消耗的资源过多,比如 I/O 与数据库的连接等。
注意事项:getInstance() 方法中需要使用同步锁 synchronized (Singleton.class) 防止多线程同时进入造成 instance 被多次实例化。
2.应用:
1.单例模式【数据库连接池】 为什么用单例模式? 答: 永远保证就一个实例。 使用方式有 4 种: 第1种: 文件的形式。(导入模块) 第2种: 类的形式。 第3种: 基于 __new__ 方式实现单例模式 第4种: 基于 __metaclass__ 方式实习单例模式 基于类方法: 第一种方式: 无法支持多线程。即使能执行,也是偶然的。 第二种方式: 加把锁,就能支持多线程。 第三种方式: 判断有没有instance存在
基于类方法:
50分单例实例:
class Singleton(object): @classmethod def instance(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): Singleton._instance = Singleton(*args, **kwargs) return Singleton._instance obj1 = Singleton.instance() print(obj1) obj2 = Singleton.instance() print(obj2)
75分:含有bug
# 单例模式:无法支持多线程情况 class Singleton(object): def __init__(self): #构造方法 import time time.sleep(1) #耗时长一些(例如:链接下数据库,或者做些其它操作,都是会造成阻塞, # 这是时候,单例模式就无效了,那怎么办? 答案: 加锁) @classmethod def instance(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): Singleton._instance = Singleton(*args, **kwargs) #这里调用了构造方法 return Singleton._instance import threading #多线程 def task(arg): obj = Singleton.instance() #实例化对象 print(obj) for i in range(10): # 创建 10 个线程 t = threading.Thread(target=task,args=[i,]) t.start()
100分:完成版
# # 单例模式:支持多线程情况 import time import threading class Singleton(object): _instance_lock = threading.Lock() #实例化一个lock对象 def __init__(self): time.sleep(1) @classmethod def instance(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): # 如果没有instance才进去,有的话就直接返回。 with Singleton._instance_lock: #当一个线程进来了,这里就会加把锁,只有一个人通过,其他人等待 if not hasattr(Singleton, "_instance"): #当第二个人进来,已经存在了就不再创建了 Singleton._instance = Singleton(*args, **kwargs) return Singleton._instance def task(arg): obj = Singleton.instance() print(obj) for i in range(10): t = threading.Thread(target=task,args=[i,]) t.start() time.sleep(20) #20秒之后,上面的进程都已经执行完,没人和下面的进程抢了 obj = Singleton.instance() #这时候,又想要获取下对象,没人来抢,就不需要枷锁了。 print(obj)
基于 __new__ 方式:
class Singleton(object): def __init__(self): print('init',self) #打印self def __new__(cls, *args, **kwargs): o = object.__new__(cls, *args, **kwargs) print('new',o) #里面 o 表示 真正创建的对象。 return o #返回值其实对应的就是下面实例化的obj对象。 #知识点: 先走 __new__方法,然后才走 __init__ 方法。 obj = Singleton() print('xxx',obj)
打印的值:
小结:
1.在构造方法执行之前,先执行 __new__ 里面的方法。
2.在 __new__ 里面的 o 就是创建的那个对象 。
3. 修改下 __new__ 方法里,加个判断: 如果有instance 就让它返回,如果没有instance 就创建。(要实现单例,还是需要加锁)
更新之后:
import time import threading class Singleton(object): _instance_lock = threading.Lock() def __init__(self): pass def __new__(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): with Singleton._instance_lock: if not hasattr(Singleton, "_instance"): Singleton._instance = object.__new__(cls, *args, **kwargs) return Singleton._instance def task(arg): obj = Singleton() print(obj) for i in range(10): t = threading.Thread(target=task,args=[i,]) t.start()
打印如下:
基于 metaclass 方式实现实例:
1. 对象是类创建的。
2. 类 是由 type 创建的。 (如何自己指定类?如下)
class SingletonType(type): pass class Foo(metaclass = SingletonType): pass
基于metaclass方法: 1,对象是由类创建,创建对象时候,类__init__方法自动执行,对象()执行类的__call__方法) 2.类 是由type创建,创建类时候,type的__init__方法自动自行,类()执行type的__call__方法(类的__new__方法,类的__init__方法)
# 第0步: 执行type的 __init__ 方法【类是type的对象】
class Foo:
def __init__(self):
pass
def __call__(self, *args, **kwargs):
pass
# 第1步: 执行type的 __call__ 方法
# 1.1 调用 Foo类(是type的对象)的 __new__方法,用于创建对象。
# 1.2 调用 Foo类(是type的对象)的 __init__方法,用于对对象初始化。
obj = Foo()
# 第2步:执行Foodef __call__ 方法
obj()
为了伪造以上现象的发生,如下举了个例子:
class SingletonType(type): def __init__(self,*args,**kwargs): super(SingletonType,self).__init__(*args,**kwargs) def __call__(cls, *args, **kwargs): obj = cls.__new__(cls,*args, **kwargs) #创建一个对象 cls.__init__(obj,*args, **kwargs) # Foo.__init__(obj) return obj class Foo(metaclass=SingletonType): def __init__(self,name): self.name = name def __new__(cls, *args, **kwargs): #cls 当前的类 return object.__new__(cls, *args, **kwargs) obj = Foo('name')
最后:
import threading class SingletonType(type): _instance_lock = threading.Lock() def __call__(cls, *args, **kwargs): #这里的cls代指: 下面的Foo类 if not hasattr(cls, "_instance"): with SingletonType._instance_lock: if not hasattr(cls, "_instance"): cls._instance = super(SingletonType,cls).__call__(*args, **kwargs) return cls._instance class Foo(metaclass=SingletonType): def __init__(self,name): self.name = name obj1 = Foo('name') obj2 = Foo('name') print(obj1,obj2)
打印如下:
也是实现 单例模式。 (这种方法比较简单。)
总结:
1.单例模式在哪用过:
1.stark 组件里面用过, (基于文件)
2.数据库连接池里面用过,(基于文件,也可基于类)
2.数据库连接池只创建了一次,我们后面拿的都是它的 conn,永远创建一份pool(池),我们就可以用单例模式。
应用单例模式:(连接池)
app.py
from pool import SingletonDBPool def run(): pool = SingletonDBPool() con = pool.connect() # xxxxxx con.close() if __name__ == '__main__': run()
pool.py
import pymysql import threading from DBUtils.PooledDB import PooledDB class SingletonDBPool(object): _instance_lock = threading.Lock() def __init__(self): self.pool = PooledDB( creator=pymysql, # 使用链接数据库的模块 maxconnections=6, # 连接池允许的最大连接数,0和None表示不限制连接数 mincached=2, # 初始化时,链接池中至少创建的空闲的链接,0表示不创建 maxcached=5, # 链接池中最多闲置的链接,0和None不限制 maxshared=3, # 链接池中最多共享的链接数量,0和None表示全部共享。PS: 无用,因为pymysql和MySQLdb等模块的 threadsafety都为1,所有值无论设置为多少,_maxcached永远为0,所以永远是所有链接都共享。 blocking=True, # 连接池中如果没有可用连接后,是否阻塞等待。True,等待;False,不等待然后报错 maxusage=None, # 一个链接最多被重复使用的次数,None表示无限制 setsession=[], # 开始会话前执行的命令列表。如:["set datestyle to ...", "set time zone ..."] ping=0, # ping MySQL服务端,检查是否服务可用。# 如:0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always host='127.0.0.1', port=3306, user='root', password='123', database='pooldb', charset='utf8' ) def __new__(cls, *args, **kwargs): if not hasattr(SingletonDBPool, "_instance"): with SingletonDBPool._instance_lock: if not hasattr(SingletonDBPool, "_instance"): SingletonDBPool._instance = object.__new__(cls, *args, **kwargs) return SingletonDBPool._instance def connect(self): return self.pool.connection()