5分钟用Python创建一个单例模式
单例模式(Singleton Pattern),就是整个程序有且仅有一个相同的实例。该类负责创建自己的对象,同时确保只有一个对象被创建
应用场景
-
你希望这个类只有一个且只能有一个实例,如某个初始化含登录的用户类,你不想在多处实例同一个用户仍需登录多次;; -
项目中的一些全局管理类(Manager)可以用单例来实现。
类似案例:Python logging 模块的 getLogger
# ~/Lib/logging/__init__.py
class Manager(object):
def getLogger(self, name):
"""
Get a logger with the specified name (channel name), creating it
if it doesn't yet exist. This name is a dot-separated hierarchical
name, such as "a", "a.b", "a.b.c" or similar.
If a PlaceHolder existed for the specified name [i.e. the logger
didn't exist but a child of it did], replace it with the created
logger and fix up the parent/child references which pointed to the
placeholder to now point to the logger.
"""
rv = None
if not isinstance(name, str):
raise TypeError('A logger name must be a string')
_acquireLock()
try:
if name in self.loggerDict:
rv = self.loggerDict[name]
if isinstance(rv, PlaceHolder):
ph = rv
rv = (self.loggerClass or _loggerClass)(name)
rv.manager = self
self.loggerDict[name] = rv
self._fixupChildren(ph, rv)
self._fixupParents(rv)
else:
rv = (self.loggerClass or _loggerClass)(name)
rv.manager = self
self.loggerDict[name] = rv
self._fixupParents(rv)
finally:
_releaseLock()
return rv
def getLogger(name=None):
"""
Return a logger with the specified name, creating it if necessary.
If no name is specified, return the root logger.
"""
if name:
return Logger.manager.getLogger(name)
else:
return root
不同的姿势实现单例
基于 __new__ 方式实现
不考虑线程安全的情况下实现单例模式
class SimpleSingleton:
def __init__(self, fuid=10001):
self.fuid = fuid
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "_instance"):
cls._instance = super().__new__(cls)
return cls._instance
<__main__.SimpleSingleton object at 0x00000250F3133308>
<__main__.SimpleSingleton object at 0x00000250F3133308>
<__main__.SimpleSingleton object at 0x00000250F3133308>
考虑线程安全的单例模式
上面创建单例虽然简单,但如果在 __new__ 方法中有一些耗时操作,在使用多线程时很可能会变成非单例模式,原因很简单,多个线程都进入了实例生成阶段。
import time
import threading
class SimpleSingleton:
def __init__(self, fuid=10001):
self.fuid = fuid
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "_instance"):
time.sleep(1) # 模拟耗时操作
cls._instance = super().__new__(cls)
return cls._instance
def task(arg):
obj = SimpleSingleton()
print(obj)
for i in range(3):
t = threading.Thread(target=task,args=[i,])
t.start()
<__main__.SimpleSingleton object at 0x0000020D45E85C88>
<__main__.SimpleSingleton object at 0x0000020D45E85288>
<__main__.SimpleSingleton object at 0x0000020D45E85908>
这时返回的就不是同一个实例了,应该如何避免呢?加锁
class Myclass:
_instance_lock = threading.Lock() # 支持多线程的单例模式
def __init__(self, fuid=10001):
self.fuid = fuid
print('call __init__')
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "_instance"):
with cls._instance_lock:
time.sleep(1) # 模拟耗时操作
if not hasattr(cls, "_instance"):
cls._instance = super().__new__(cls)
return cls._instance
obj1 = Myclass(10001)
obj2 = Myclass(fuid=10001)
obj3 = Myclass(fuid=10002)
print(obj1)
print(obj2)
print(obj3)
call __init__
call __init__
call __init__
<__main__.Myclass object at 0x00000197B57F0EC8>
<__main__.Myclass object at 0x00000197B57F0EC8>
<__main__.Myclass object at 0x00000197B5AB0C88>
1. 两层 if 判断(双重检测机制)
细心的朋友可能关注到上面面为什么用了两层if 判断,可以试想,两个thead执行到是否有_instance属性这个判断thead1先通过,执行下一句with cls._instance_lock,那么便直接占有了锁,之后在thread1的1秒sleep中thread2也通过了第一个if判断,而继续执行执行with cls._instance_lock语句,无法抢占锁,被阻塞,当thread1完成1秒的sleep后,并且通过第二个if,对cls._instance赋值,退出with context后,thread2才能继续执行,1 秒sleep之后再进行第二个if判断,此时不能通过了,因为thread1已经创建了一个instance那么只好退出with context,再执行return cls._instance,其实就是返回thread1创建的cls._instance
2. __init__ 仍调用了多次
还有个问题,虽然我们得到的类实例都是同一个(0x00000197B57F0EC8),但__init__ 仍被调用了多次,如果在初始化进行数据库连接等操作时,也会连接多次,这不是我们想要的, 为什么是这样?
__new__负责对象的创建,而__init__负责对象的初始化;__new__是一个类方法,而__init__是一个对象方法。
__new__是我们通过类名进行实例化对象时自动调用的,__init__是在每一次实例化对象之后调用的,__new__方法创建一个实例之后返回这个实例对象,并将其传递给__init__方法的 self 参数,也就是说__new__返回的对象在真正完成实例化前还是会调用__init__
How To:一种方案就是在 __init__调用时也做一次判断, 如下
class Myclass:
_instance_lock = threading.Lock() # 支持多线程的单例模式
_is_first_init = False
def __init__(self, fuid=10001):
if not self._is_first_init:
self.fuid = fuid
print('call __init__')
self._is_first_init = True
def __new__(cls, *args, **kwargs):
if not hasattr(cls, "_instance"):
with cls._instance_lock:
time.sleep(1) # 模拟耗时操作
if not hasattr(cls, "_instance"):
cls._instance = super().__new__(cls)
return cls._instance
显然这是不优雅的,推荐使用下方的 metaclass 或 装饰器 实现方式
基于 metaclass 方式实现 [推荐]
class SingletonMetaClass(type):
_instance_lock = threading.Lock() # 支持多线程的单例模式
_instance = {}
def __call__(cls, *args, **kwargs):
fuid = args[0] if args else kwargs.get('fuid') or 10001
if not cls._instance.get(fuid):
with cls._instance_lock:
if not cls._instance.get(fuid):
cls._instance[fuid] = super().__call__(*args, **kwargs)
return cls._instance[fuid]
class Myclass(metaclass=SingletonMetaClass):
def __init__(self, fuid=10001):
self.fuid = fuid
print('call __init__')
基于装饰器方式实现 [推荐]
def singleton_wrapper(cls, *args, **kwargs):
"""定义一个单例装饰器"""
instance = {}
def wrapper(*args, **kwargs):
if cls not in instance:
instance[cls] = cls(*args, **kwargs)
return instance[cls]
return wrapper
@singleton_wrapper
class Myclass:
def __init__(self, fuid=10001):
self.fuid = fuid
print('call __init__')
