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__')