Skip to main content

单例模式

单例的定义

单例模式就是保证某个实例在项目的整个生命周期中只存在一个,在项目的任意位置使用,都是同一个实例。

必要性

  1. 全局只有一个接入点,可以更好地进行数据同步控制,避免多重占用;
  2. 由于单例模式要求在全局内只有一个实例,因而可以节省比较多的内存空间;
  3. 单例可长驻内存,减少系统开销。

和其他设计模式一样,单例模式有一定的适用场景,但同时它也会给我们带来一些问题。

  1. 由于单例对象是全局共享,所以其状态维护需要特别小心。一处修改,全局都会受到影响。
  2. 单例对象没有抽象层,扩展不便。
  3. 赋于了单例以太多的职责,某种程度上违反单一职责原则;
  4. 单例模式是并发协作软件模块中需要最先完成的,因而其不利于测试;
  5. 单例模式在某种情况下会导致“资源瓶颈”。

常用方案

python因为存在模块化,所以单例模式并不需要特别去实现,模块化就是最天然的单例

官方实践文档

单线程:

  • 使用import以模块引入并使用

  • 使用函数装饰器实现单例

  • 使用类装饰器实现单例

  • 使用 __new__ 关键字实现单例

  • 使用 metaclass 实现单例

多线程:

  • 通过加锁配合上述功能

    import threading

    # 实现一个给函数加锁的装饰器
    def synchronized(func):

    func.__lock__ = threading.Lock()

    def lock_func(*args, **kwargs):
    with func.__lock__:
    return func(*args, **kwargs)
    return lock_func
    # 然后在实例化对象的函数上,使用这个装饰函数。
    import time
    import threading

    class User:
    _instance = None

    @synchronized
    def __new__(cls, *args, **kwargs):
    if not cls._instance:
    time.sleep(1)
    cls._instance = super().__new__(cls)
    return cls._instance

    def __init__(self, name):
    self.name = name

    def task():
    u = User("wangbm")
    print(u)

    for i in range(10):
    t = threading.Thread(target=task)
    t.start()

判断是否单例

  • 通过id()内置函数,判断两个对象实例

实现案例

原生模块化(首选推荐)

使用import的天然机制,实现真正的单例模式,其他所有实现方法均不推荐

  • 定义
# test.py
class My_Singleton(object):
def foo(self):
pass

# 在模块内创建一个实例
# 外部通过引入模块内的实例来达到单例
# from xxx import my_singleton
my_singleton = My_Singleton()
  • 使用
from test import my_singleton

函数装饰器

  • 定义
# 单例装饰器定义
def singleton(cls):
_instance = {}

def inner():
if cls not in _instance:
_instance[cls] = cls()
return _instance[cls]

return inner
  • 使用
@ singleton
class Test():
def __init__(self):
pass

cls1 = Test()
cls2 = Test()
print(id(cls1) == id(cls2)) # True

类装饰器

实现原理:

​ 建立一个装饰器类,该类通过 __call__方法来劫持对象

  • 定义
class Singleton(object):
def __init__(self, cls):
self._cls = cls
self._instance = {}
def __call__(self):
if self._cls not in self._instance:
self._instance[self._cls] = self._cls()
return self._instance[self._cls]
  • 使用方法1:装饰器
@Singleton
class Cls2(object):
def __init__(self):
pass

cls1 = Cls2()
cls2 = Cls2()
print(id(cls1) == id(cls2)) # True

  • 使用方法2:继承
class Cls3():
pass

Cls3 = Singleton(Cls3)
cls3 = Cls3()
cls4 = Cls3()
print(id(cls3) == id(cls4))

使用 __new__ 关键字实现(伪单例,不推荐)

这种方式被很多文章引用,但是实际上,这并不是单例,一下代码就能说明

image-20221006014938490

采用__new__()创建的对象只能确保是同一内存地址的同一个对象,但是实例化__init__()会将导致对象重新初始化,并不是返回原来的对象

原文链接

因为每次创建实例时,对象都会调用 __new__ 方法,然后判断并重新返回对象

虽然能达到单例模式的作用,但严格意更接近与覆盖赋值,用已存在的对象覆盖新对象,开销比真实的单例模式大一点

  • 定义
class Single(object):
_instance = None
def __new__(cls, *args, **kw):
if cls._instance is None:
cls._instance = object.__new__(cls)
return cls._instance

def __init__(self):
pass

single1 = Single()
single2 = Single()
print(id(single1) == id(single2))

在理解到 new 的应用后,理解单例就不难了,这里使用了

_instance = None

来存放实例,如果 _instance 为 None,则新建实例,否则直接返回 _instance 存放的实例。

这种写法存在有两个问题:

  • 实际类在初始化的时候,无法通过__init__()这个方法传递参数,会显示错误

    image-20221006011742957

    TypeError: object.new() takes exactly one argument (the type to instantiate)

    object.__new__(cls.*args,**kw)的参数去掉,即可不再报错

  • 多个线程实例化时,可能会创建多个示例的情况,因为存在多个线程同时判断None的情况,从而在同一时间单位内创建了不同的实例,解决方法:

    import threading

    # 同步锁
    def synchronous_lock(func):
    def wrapper(*args, **kwargs):
    with threading.Lock():
    return func(*args, **kwargs)
    return wrapper

    class Singleton(object):
    instance = None

    @synchronous_lock
    def __new__(cls, *args, **kwargs):
    if cls.instance is None:
    cls.instance = object.__new__(cls, *args, **kwargs)
    return cls.instance

    通过在 __new__()中加入线程锁的装饰器,将实例化的过程同步化,就不会出现上述问题,不过也将单例化更加复杂:

# 测试代码
def worker():
s = Singleton()
print(id(s))

def test():
task = []
for i in range(10):
t = threading.Thread(target=worker)
task.append(t)
for i in task:
i.start()
for i in task:
i.join()

test()

完整方案

def singleton(cls):
cls.__new_original__ = cls.__new__

@functools.wraps(cls.__new__)
def singleton_new(cls, *args, **kwargs):
it = cls.__dict__.get('__it__')
if it is not None:
return it

cls.__it__ = it = cls.__new_original__(cls, *args, **kwargs)
it.__init_original__(*args, **kwargs)
return it

cls.__new__ = singleton_new
cls.__init_original__ = cls.__init__
cls.__init__ = object.__init__
return cls

@singleton
class Foo(object):
def __new__(cls, *args, **kwargs):
cls.x = 10
return object.__new__(cls)

def __init__(self, x, y):
assert self.x == 10
self.x = x
self.y = y

上述代码中定义了singleton类装饰器,装饰器在预编译时就会执行,利用这个特性,singleton类装饰器中替换了类原本的__new____init__方法,使用singleton_new方法进行类的实例化,在singleton_new方法中,先判断类的属性中是否存在__it__属性,以此来判断是否要创建新的实例,如果要创建,则调用类原本的__new__方法完成实例化并调用原本的__init__方法将参数传递给当前类,从而完成单例模式的目的。

这种方法让单例类可以接受对应的参数但面对多线程同时实例化还是可能会出现多个实例,此时加上线程同步锁则可。

def singleton(cls):
cls.__new_original__ = cls.__new__
@functools.wraps(cls.__new__)
def singleton_new(cls, *args, **kwargs):
# 同步锁
with threading.Lock():
it = cls.__dict__.get('__it__')
if it is not None:
return it

cls.__it__ = it = cls.__new_original__(cls, *args, **kwargs)
it.__init_original__(*args, **kwargs)
return it

cls.__new__ = singleton_new
cls.__init_original__ = cls.__init__
cls.__init__ = object.__init__
return cls

是否添加同步锁

如果一个项目不需要使用线程相关机制,只是在单例化这里使用了线程锁,这其实不是必要的,它会拖慢项目的运行速度。

阅读CPython线程模块相关的源码,你会发现,Python一开始时并没有初始化线程相关的环境,只有当你使用theading库相关功能时,才会调用PyEval_InitThreads方法初始化多线程相关的环境,多线程环境会启动GIL锁相关的逻辑,这会影响Python程序运行速度。很多简单的Python程序并不需要使用多线程,此时不需要初始化线程相关的环境,Python程序在没有GIL锁的情况下会运行的更快。

如果你的项目中不会涉及多线程操作,那么就没有使用有同步锁来实现单例模式。

使用 metaclass 实现

同样,我们在类的创建时进行干预,从而达到实现单例的目的。

在实现单例之前,需要了解使用 type 创造类的方法,代码如下:

def func(self):
print("do sth")

Klass = type("Klass", (), {"func": func})

c = Klass()
c.func()

以上,我们使用 type 创造了一个类出来。这里的知识是 mataclass 实现单例的基础。

class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]

class Cls4(metaclass=Singleton):
pass

cls1 = Cls4()
cls2 = Cls4()
print(id(cls1) == id(cls2))

这里,我们将 metaclass 指向 Singleton 类,让 Singleton 中的 type 来创造新的 Cls4 实例。

单例模式的坑