目录:
- 单例模式在类中实现
- 装饰器批量装饰实现单例模式 ,且不丢失类型提示
- 限制实例个数
1.重写__new__方法实现多线程情况下的单例模式
用new方法实现单例模式
import time, threadingclass Singleton:"""单例模式————最多只允许创建一个该类实例"""_lock = threading.Lock()_instance = Nonedef __new__(cls, a, b, *args, **kwargs):time.sleep(1) # 模拟高并发情况with cls._lock:# print("进入lock区域")if not cls._instance:cls._instance = object.__new__(cls)# cls._instance = super().__new__(cls) # 与上一条作用相同# cls._instance.a = areturn cls._instance# S = Singleton(1, 2)
# print(S)def task(a, b):s = Singleton(a, b)print(s)for i in range(5):t = threading.Thread(target=task, args=(2, 3))t.start()
运行始终为单一实例:
<__main__.Singleton object at 0x0000029E71D3B070><__main__.Singleton object at 0x0000029E71D3B070>
<__main__.Singleton object at 0x0000029E71D3B070><__main__.Singleton object at 0x0000029E71D3B070><__main__.Singleton object at 0x0000029E71D3B070>
2.使用装饰器实现多线程情况下的单例模式
想要批量给多个类实现单例模式,自然少不了装饰类的装饰器
先介绍第一种方法,线程锁的方式和前面的稍有不一样,没有使用类的内部属性记录实例对象,而是使用了字典键唯一的特性,将类对象Example作为键,实例化后的对象作为值
2.1 第一种实现方式(不推荐使用)
错误示例:
import time
from functools import wraps
import threadingdef singleton_dec(class_):map_ = {} # 定义一个字典,类本身作为键,实例化后的对象作为其相应的值@wraps(class_) # (顺便复习一下装饰器,将被装饰的函数改回原来的属性和名字)def singleton(*args):# print(class_.__name__)# print(class_.foo.__name__)if class_ not in map_: # 如果类没有在字典就实例化time.sleep(1) # 没有加锁的情况下模拟多线程的情况map_[class_] = class_(*args) # 实例化类方法并存到字典中return map_[class_] # 字典中有这个类就不实例化return singleton@singleton_dec
class EXAMPLE:def __init__(self, param='abc'):self.param = paramdef foo(self):pass# E1 = EXAMPLE('123')
# print(E1.param)
# print(E1)
# E2 = EXAMPLE('456') # 由于EXAMPLE已经实例化了(在map_中),这一步并没有对EXAMPLE做出任何改变
# print(E2.param) # 输出的依然是123
# print(E2)def task():e = EXAMPLE()print(e)for i in range(5):t = threading.Thread(target=task)t.start()
加上sleep模拟多线程的执行结果:
<__main__.EXAMPLE object at 0x000001BD894A7128>
<__main__.EXAMPLE object at 0x000001BD8949E710>
<__main__.EXAMPLE object at 0x000001BD894A1DD8>
<__main__.EXAMPLE object at 0x000001BD8949EB00>
<__main__.EXAMPLE object at 0x000001BD8949E710>
可以看到在并发情况很高的时候并没有实现单例化
因为方法二没有像方法一一样设置线程锁,所以这里会出现创建多个实例的情况
想要实现单例,也需要像方法一中一样加上线程锁
不过是给字典方法加上一个线程锁:
示例:
# -*- coding: utf-8 -*-
import threading, time
from functools import wrapsclass DictWithLock(dict):lock = threading.Lock()def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)def singleton_decorator(callable_obj):dl = DictWithLock()@wraps(callable_obj)def inner(*args, **kwargs):# print('测试', *args, **kwargs)time.sleep(1)with dl.lock: # 一次只能有一个线程使用DictWithLock的lock属性# print("进入lock区域")if not dl.get("_instance"):dl["_instance"] = callable_obj(*args, **kwargs)return dl["_instance"]return inner@singleton_decorator
class Test:def __init__(self, a=0, b=1):print('Test.test', a, b)def task(a, b):t = Test(a, b)print(t) # t.a时不会有提示# t0 = Test(2, 3)
# print(t0)for i in range(5):tsk = threading.Thread(target=task, args=(2, 3))tsk.start()
输出结果:
Test.test 2 3
<__main__.Test object at 0x00000292250AB070>
<__main__.Test object at 0x00000292250AB070><__main__.Test object at 0x00000292250AB070><__main__.Test object at 0x00000292250AB070><__main__.Test object at 0x00000292250AB070>已结束,退出代码 0
线程各自异步执行,但始终只会创建一个Example的实例
这种方式是按照一般实现装饰器的方式实现,不过这种装饰器有明显的缺点,被装饰过的类不会有类型提示
于是祭出第二种类的装饰器写法
2.2 第二种实现方式
被装饰过的类不会丢失类型提示
# -*- coding: utf-8 -*-
import threadingdef Singleton(cls):# print("log singleton")cls._instance = Nonecls._lock = threading.Lock()def __new__(*args, **kwargs): # print(f"log new", args, kwargs) # 实际上args中也包含cls和相关的参数with cls._lock:if not cls._instance:cls._instance = object.__new__(cls)return cls._instancecls.__new__ = __new__return cls@Singleton
class T:def __init__(self, a):self.a = adef t1():t = T(1)tt = T(2)print(t, t.a) # t.a时有提示print(tt, t.a)def t11():def task(a):s = T(a)print(s)for i in range(5):t = threading.Thread(target=task, args=(2, ))t.start()if __name__ == '__main__':# t1()t11()
输出:
<__main__.T object at 0x000001EFE5977850>
<__main__.T object at 0x000001EFE5977850>
<__main__.T object at 0x000001EFE5977850>
<__main__.T object at 0x000001EFE5977850>
<__main__.T object at 0x000001EFE5977850>
限制实例个数的模式————limit实例模式
__new__()方法实现最多创建3个实例的模式:
class LimitedInstances:"""限制实例个数的模式————limit实例模式"""_instances = []limit = 3def __new__(cls, *args, **kwargs):if not len(cls._instances) < cls.limit: # 若个数达到limit说明已满raise Exception(f"Can not create instance more than {cls.limit}.")instance = super().__new__(cls)cls._instances.append(instance)return instancedef __delete__(self):self._instances.remove(self)LI1 = LimitedInstances(1, 2)
LI2 = LimitedInstances(2, 2)LI3 = LimitedInstances(3, 2)
LI3.__delete__()
LI4 = LimitedInstances(4, 2)
print(f"删除一个又加进来一个,还是3个\n", LimitedInstances._instances)
# LI5 = LimitedInstances(5, 2) # 超过3个会报错
运行结果,限制最多3个实例:
删除一个又加进来一个,还是3个实例[<__main__.LimitedInstances object at 0x00000278A03BDE20>, <__main__.LimitedInstances object at 0x00000278A03BDDC0>, <__main__.LimitedInstances object at 0x00000278A03BD520>]
模拟高并发情况下,还是可能会出现超过3个实例的情况:
import time, threadingclass LimitedInstances:"""限制实例个数的模式————limit实例模式"""_instances = []limit = 3def __new__(cls, *args, **kwargs):if not len(cls._instances) < cls.limit: # 若个数达到limit说明已满raise Exception(f"Can not create instance more than {cls.limit}.")instance = super().__new__(cls)time.sleep(1) # 模拟高并发情况cls._instances.append(instance)return instancedef __delete__(self):self._instances.remove(self)def task():LI = LimitedInstances()print(LI)def task():LI = LimitedInstances()print(LI)for i in range(6):t = threading.Thread(target=task, args=())t.start()
time.sleep(2)
print(len(LimitedInstances._instances), LimitedInstances._instances)
运行结果,出现了6个不同的实例的情况:
<__main__.LimitedInstances object at 0x0000027066EC8DF0><__main__.LimitedInstances object at 0x0000027066EED370><__main__.LimitedInstances object at 0x0000027066EC8BB0>
<__main__.LimitedInstances object at 0x0000027066EED0D0>
<__main__.LimitedInstances object at 0x0000027066EC8610>
<__main__.LimitedInstances object at 0x0000027066EED610>
6 [<__main__.LimitedInstances object at 0x0000027066EC8DF0>, <__main__.LimitedInstances object at 0x0000027066EED370>, <__main__.LimitedInstances object at 0x0000027066EC8BB0>, <__main__.LimitedInstances object at 0x0000027066EED0D0>, <__main__.LimitedInstances object at 0x0000027066EC8610>, <__main__.LimitedInstances object at 0x0000027066EED610>]
解决方式,同之前,加上线程锁:
import time, threadingclass LimitedInstances:"""限制实例个数的模式————limit实例模式"""_instances = []limit = 3_lock = threading.Lock()def __new__(cls, *args, **kwargs):with cls._lock: # 一次只能一个线程执行实例个数判断和实例存入if not len(cls._instances) < cls.limit: # 若个数达到limit说明已满raise Exception(f"Can not create instance more than {cls.limit}.")instance = super().__new__(cls)time.sleep(1) # 模拟高并发情况cls._instances.append(instance)return instancedef __delete__(self):self._instances.remove(self)def task():LI = LimitedInstances()print(LI)for i in range(6):t = threading.Thread(target=task, args=())t.start()
# time.sleep(2)
# print(len(LimitedInstances._instances), LimitedInstances._instances)
运行时,创建超过了3个实例会报错