对于很多开发人员来说,单例模式算是比较简单常用、也是最早接触的设计模式了,仔细研究起来单例模式似乎又不像看起来那么简单。我们知道单例模式适用于提供全局唯一访问点,频繁需要创建及销毁对象等场合,的确方便了项目开发,但是单例模式本身也有一定的局限性,如果滥用则会给后续软件框架的扩展和维护带来隐患。
单例模式的实现有很多种,应用场合也各有不同,但必须保证实例唯一,如果是多线程环境则必须保证线程安全。python本身有很多内置特性可以用来实现单例模式的效果,理清每种单例模式的实现原理、优缺点及使用场合才能更好地用其长避其短。
定义
单例模式: 保证一个类只有一个实例,并提供一个访问该实例的全局访问点
应用场景
单例模式适用于很多场合,各个产品也有不少单例模式的实现例子,比如windows的回收站,网站计数器等。此处仅列举跟软件应用开发强相关的常用场景:
- 用于保存全局状态,如全局的配置文件
- 用于避免频繁创建及销毁对象,如数据库连接池
- 用于多线程之间共享对象资源,如线程池
优缺点
我们知道硬币都有两面,任何一种设计模式既有它的优势,也会有它的局限性。所以非常有必要清楚每种设计模式的优缺点扬长避短。
- 优点
- 提供了全局访问点,方便统一管理
- 避免了对同一资源的访问竞争及操作冲突
- 减小了开销,尤其是需要频繁创建和销毁的对象
- 类可以根据需要方便灵活地控制实例化过程
- 缺点
- 不适用于变化的对象,不可以根据不同的应用场景创建不同的对象
- 单例模式没有抽象层,不易扩展及解耦
- 承担的职责过多,违反单一职责原则
- 进程被杀时容易引发内存泄漏、及资源未释放的问题
- 单例模式中创建新对象的方式和项目开发中默认的创建新对象的方式不一致,这会隐性增加开发成本
单例模式的校验
实例唯一
单例模式的出发点就是通过唯一的实例提供全局访问点,当然在多线程环境下还有线程安全的要求。这里先说一下如何检验单例模式生成的实例是否唯一,线程安全的话题在下面。
python中的每个对象都包含三个要素:
- id: 唯一标识一个对象
- type: 标识对象的类型
- value: 对象的值
a is b: 用来判断对象a是否就是对象b,是通过id来判断的
a == b: 用来判断对象a的值是否和对象b的值相等,是通过value来判断的
所以如果运行中的单例模式每次都生成ID相同的实例,也就验证了该单例模式的实例唯一性。
线程安全
线程安全简单地说就是多个线程同时执行某个程序时不会发生写冲突。
要想使程序在多线程环境下依旧运行正确,就必须保证同一时刻只有一个线程对共享数据进行存取,多线程环境下通常使用加锁来保证存取操作的唯一性和排他性。线程安全一直是并发领域的一个难点,不管对于静态语言还是动态语言都有很多内置的、常用的collections是非线程安全的,比如Java的HashMap本身是非线程安全的,使用不当的话会造成严重的性能问题,但是由于其应用非常广泛,不少人对它的非线程安全性并不敏感,所以在多线程环境中一定要使用其对应的线程安全版本;再比如python内置的list,dict,set,iterator都不是线程安全的,需要我们自己加锁进行线程同步。
就单例模式的线程安全来说,有两个认识上的误区,一是有人觉得单例模式本身只能生成唯一实例,ID都是同一个,既然是独苗应该不会再有线程安全的问题,二是python本身提供的GIL机制本身就是保证同一时刻只有一个线程对共享数据进行存取的,所以python内置的GIL机制已经从原子操作方面保证线程安全了。事实真的是这样吗?这就需要先进行原理分析、再进行数据测试来检验了。
误区一:实例唯一与线程安全
实例唯一性和线程安全性是两个不同的范畴,实例唯一是保证只有一个全访问点,这在单线程中当然是没有问题的,但是在多线程环境中全局访问点作为共享资源本身就是资源竞争的对象,有多个线程都想获得这个唯一实例的操作权限。
我们知道软件项目中有很多实现都是面向事务的,事务本身是由多个任务组合在一起的,要想保证事务的安全性就要求操作成功时提交整个事务,操作失败时回滚整个事务,对于单例模式的唯一实例也是如此,试想如果线程A在某个时间片段获得该唯一实例的操作权限,执行到一半任务时,因为时间片到、中断、函数调用等原因交出操作权限,线程B获得该唯一实例的控制权,此时如果线程A和线程B之间相互独立、没有共享存取资源的话当然没问题,如果有共享资源则很容易出现数据或事务污染,也就是非线程安全,无法保证程序正确运行。所以即使是唯一实例也需要加锁保证对线程安全敏感的事务能够正确提交或整体回滚。
误区二:GIL与原子操作
GIL的官方解释是:全局解释器锁。每个线程在执行到过程中都需要先获取GIL,保证同一时刻只有一个线程可以执行代码。
这听起来很线程安全,但是问题是GIL的线程安全是基于原子操作的,而我们实际的执行语句或执行代码是由一个或多个原子操作组成的,如果每条语句或者每块代码整体上都是一个原子操作这当然也没有问题,但事实上这是不可能的,有些语句比如x = x + 1本身就表示tmp = x + 1及x = tmp(tmp表示中间变量)两个原子操作,也就是说这条语句在GIL机制下无法保证线程安全。如果在多线程下执行 x = x + 1这条语句,不对其加锁进行同步的话无法保证最终计算结果的正确性。
非线程安全的单例实现
上面说的都是理论上的解释,最好有实际测试数据佐证看起来更清楚、直白,更有说服力。先看一下未加锁的单例实现及其在多线程下的执行结果,因为单纯的单例实现无法测试出该单例模式是否是线程安全的,所以我们给该单例模式加上相应的数据计算,如果数据计算最终结果与预期一致就表示线程安全,如果不一致就表示多线程环境下程序未能正常执行,该单例实现是非线程安全的,当然为了避免偶然性的发生,我们可以多次执行同一个测试程序。
要添加的数据计算是
zero += 1
zero -= 1
从理论上说,先+1再-1,无论执行多少次,zero变量的值都应该是0。将这两个语句封装到单例中,并加大循环次数(数据量大更容易出现线程安全性问题,程序执行时间可能会多花几秒种),保存成mysingleton.py:
class Singleton(object):
def __init__(self):
self.zero = 0 def __new__(cls, *args, **kw):
if not hasattr(cls, "_instance"):
cls._instance = super(Singleton, cls).__new__(cls, *args, **kw) return cls._instance def change_zero(self):
for i in range(10000000):
self.zero += 1
self.zero -= 1
封装校验类先检验但是是否唯一,再检验是否线程安全,并多次执行避免偶然性:
from mysingleton import Singleton
import threading class SingletonCheck(object):
def __init__(self):
self.task = Singleton() def action(self):
obj = Singleton()
print ("Created Object: {}".format(obj)) def only_instance_test(self):
for i in range(10):
t = threading.Thread(target=self.action)
t.start()
t.join() def thread_safety_test(self):
t1 = threading.Thread(target=self.task.change_zero)
t2 = threading.Thread(target=self.task.change_zero)
t3 = threading.Thread(target=self.task.change_zero)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join() print ("期望的zero值: 0, 实际的zero值: {}\n该实例的实现是否是线程安全的: {}".format(self.task.zero, 0 == self.task.zero)) if __name__ == "__main__":
singleton_check = SingletonCheck()
print ("开始验证该单例模式生成的实例ID是否唯一...")
singleton_check.only_instance_test()
print ("结束验证该单例模式生成的实例ID是否唯一...")
print ("=" * 64)
print ("开始验证该单例模式是否线程安全...")
singleton_check.thread_safety_test()
print ("结束验证该单例模式是否线程安全...") # 执行结果
开始验证该单例模式生成的实例ID是否唯一...
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
Created Object: <__main__.Singleton object at 0x01EE5D50>
结束验证该单例模式生成的实例ID是否唯一...
================================================================
开始验证该单例模式是否线程安全...
期望的zero值: 0, 实际的zero值: -9
该实例的实现是否是线程安全的: False
结束验证该单例模式是否线程安全...
>>>
从测试结果来看该单例模式实现只能保证实例唯一,并不能保证线程安全。
线程安全的单例实现
保证单例实现的线程安全需要加锁进行同步,python提供了两种锁threading.Lock() & threading.RLock()实现线程同步,各自有两种加锁方式with func.__lock__和lock_inst.acquire() & lock_inst.release().插桩代码。
threading.Lock() & threading.RLock()
threading.Lock():多线程中基本的锁对象,同一线程中只能一次acquire,其余的锁请求只能等锁释放后才能获取。使用过程中可能造成迭代死锁
threading.RLock():可重入锁,同一线程中可以多次acquire,acquire()和release()必须成对出现。使用过程中不会造成迭代死锁
threading.RLock()可以看成是threading.Lock()的改进版。
方式一:装饰器@synchronized实现同步操作
def synchronized(func): func.__lock__ = threading.RLock() def lock_func(*args, **kwargs):
with func.__lock__:
return func(*args, **kwargs)
return lock_func
在需要同步的方法前添加@synchronized,以上面的单例实现为例。修改后线程安全的单例模式实现,同样保存在mysingleton.py中:
def synchronized(func):
func.__lock__ = threading.RLock() def lock_func(*args, **kwargs):
with func.__lock__:
return func(*args, **kwargs)
return lock_func class Singleton(object):
def __init__(self):
self.zero = 0
@synchronized
def __new__(cls, *args, **kw):
if not hasattr(cls, "_instance"):
cls._instance = super(Singleton, cls).__new__(cls, *args, **kw) return cls._instance @synchronized
def change_zero(self):
for i in range(10000000):
self.zero += 1
self.zero -= 1
同一个封装校验类检验是否线程安全,并多次执行避免偶然性。
from mysingleton import Singleton
import threading class SingletonCheck(object):
def __init__(self):
self.task = Singleton() def action(self):
obj = Singleton()
print ("Created Object: {}".format(obj)) def only_instance_test(self):
for i in range(10):
t = threading.Thread(target=self.action)
t.start()
t.join() def thread_safety_test(self):
t1 = threading.Thread(target=self.task.change_zero)
t2 = threading.Thread(target=self.task.change_zero)
t3 = threading.Thread(target=self.task.change_zero)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join() print ("期望的zero值: 0, 实际的zero值: {}\n该实例的实现是否是线程安全的: {}".format(self.task.zero, 0 == self.task.zero)) if __name__ == "__main__":
singleton_check = SingletonCheck()
print ("开始验证该单例模式生成的实例ID是否唯一...")
singleton_check.only_instance_test()
print ("结束验证该单例模式生成的实例ID是否唯一...")
print ("=" * 64)
print ("开始验证该单例模式是否线程安全...")
singleton_check.thread_safety_test()
print ("结束验证该单例模式是否线程安全...") # 执行结果
开始验证该单例模式生成的实例ID是否唯一...
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
Created Object: <__main__.Singleton object at 0x01E65D10>
结束验证该单例模式生成的实例ID是否唯一...
================================================================
开始验证该单例模式是否线程安全...
期望的zero值: 0, 实际的zero值: 0
该实例的实现是否是线程安全的: True
结束验证该单例模式是否线程安全...
方式二:需同步代码的前后插入lock_inst.acquire() & lock_inst.release()
class Singleton(object):
__rlock = threading.RLock() def __init__(self):
self.zero = 0 def __new__(cls, *args, **kw):
Singleton.__rlock.acquire()
if not hasattr(cls, "_instance"):
cls._instance = super(Singleton, cls).__new__(cls, *args, **kw)
Singleton.__rlock.release()
return cls._instance def change_zero(self):
for i in range(10000000):
Singleton.__rlock.acquire()
self.zero += 1
self.zero -= 1
Singleton.__rlock.release()
校验过程可参考方式一。
注意事项
- 单例应该只用来保存全局状态,不应该和任何小于程序完整生命周期的作用域绑定
- 不能用反射、序列化、克隆等破坏单例唯一性的方式创建单例,否则只会实例化另一个新对象
- 单例对象一旦创建就会永久驻留内存直到程序终止,过多的单例会增大内存消耗
- 多线程环境下使用单例模式一定要保证线程安全
单例模式的实现
python本身是一门非常灵活的语言,内置许多特性,所以有很多方式实现单例模式,或者达到实例唯一的效果。如下重点介绍python内置的哪些特性可以实现单例模式,所以选择单线程下的实现较为简便,如果需要相应修改为多线程下安全的单例模式,请参考【单例模式的校验】中【线程安全】部分给需要同步的代码加锁。
1. 重写__new__()
原理分析
__new__()方法在python中属于内置函数用于创建类实例,通过重载__new__()方法可以在创建实例过程中自定义我们需要的功能。__new__()和__init__()是刚接触python是比较容易混淆的两个方法,二者的区别是__new__()用于创建实例,__init__()用于实例创建后的初始化工作,这一点具备编程基础的人应该好理解。
__new__()方法至少要有一个参数cls, 参数cls表示当前正在实例化的类。要想得到当前类的实例,应当在当前类的__new__()方法中调用父类的__new__()方法(即super(Singleton, cls).__new__(cls, *args, **kw)),如果父类是object的话,前面括号里的内容可以直接写成object.__new__(cls, *args, **kw)
使用__new__()方法创建单例过程:先判断当前类是否已存在类变量_instance,如果存在直接return该类变量,如果不存在则生成一个Singleton实例,再赋值给_instance,然后return该类变量。
单线程实现
class Singleton(object):
def __new__(cls, *args, **kw):
if not hasattr(cls, "_instance"):
cls._instance = super(Singleton, cls).__new__(cls, *args, **kw)
return cls._instance
2. 重写__call__()
原理分析
在python中,函数其实也是对象,所有的函数都是可调用对象。python中有个内置方法__call__(),如果实现了它,一个类实例也可以变成可调用对象,相当于重载了()运算符。 __call__()的作用是使实例能够像函数一样被调用,这是一件比较有意思的事,通过Singleton = Singleton()就可以实现自循环的单例调用,同时不影响实例本身的生命周期,如__new__()和__init__()的过程。
单线程实现
class Singleton(object):
def test(self):
print('Singleton Test') def __call__(self, *args, **kwargs):
return self Singleton = Singleton()
3. @staticmethod特性
原理分析
单例模式的实质就是提供实例化的唯一通道,实例化的途径是首先通过自身类的__new__来实例化及__init__进行初始化,如果自身类的实例化被禁止,则可以层层上溯同父类.__new__()及__init__()进行实例化。
在自身类的__init__()中抛出语法错误禁止实例化,同时在@staticmethod修改的静态方法中通过调用当前类的父类.__new__()生成实例,然后把该实例赋值给当前类的类变量_instance,即可以实现实例化的唯一通道。
这种方式可以实现单例模式,但是强制__init__()抛出语法错误的做法比较强制,有点粗暴干涉的意味,不太推荐此种方式,但是知道这种原理就可以了。
单线程实现
class Singleton(object):
_instance = None def __init__(self):
raise SyntaxError('instantiation error, please use get_instance()') @staticmethod
def get_instance():
if Singleton._instance is None:
Singleton._instance = object.__new__(Singleton)
return Singleton._instance
4. @classmethod特性
原理分析
使用@classmethod特性实现单例模式的原理跟使用@staticmethod实现单例模式的原理颇为相似,这里主要介绍下@staticmethod和@classmethod的区别。
- 传递参数: @classmethod修饰的方法必须使用类对象作为第一个参数,@staticmethod修饰的方法则不需要传递任何参数
- 调用方式: 二者修饰的方法都可以通过类名/实例对象来调用
- 调用对象:
- @classmethod修饰的方法持有cls参数,可以通过cls.xxx的方式调用类的变量、方法、实例等属性,避免硬编码;
- @staticmethod修饰的方法中如果想调用类的属性(变量,方法,实例等)只能通过类名.属性名(显式硬编码)的方式调用
- 使用场景:
- @classmethod: 用在需要访问当前类属性的方法,常作为当前类构造函数的补充
- @staticmethod: 确认当前类的某个方法不会涉及到与当前类属性有关的操作时,可以考虑将该方法定义为当前类的staticmethod
- 子类继承:
- 有子类继承时,调用@classmethod修饰的方法中自动传入的类变量cls是子类,而非父类,
- 有子类继承时,如果@staticmethod修饰的方法中包含显式的类名引用,则子类中不会自动替换为其子类名
单线程实现
class Singleton(object):
_instance = None def __init__(self):
raise SyntaxError('instantiation error, please use get_instance()') @classmethod
def get_instance(cls):
if Singleton._instance is None:
Singleton._instance = object.__new__(Singleton)
return Singleton._instance
5. 类属性
原理分析
python中的属性是一个比较宽泛的概念,比java中的属性范围大得多,也不同于python内置的@property,虽然property翻译过来也是属性的意思,但是python的属性和@property完全不是一回事,类中的变量、方法、实例,实例中的变量都可以成为python的属性。python属于动态语言,很多特性与静态语言不同,所以不能把静态语言的概念和规则套到python上来。
python中对于属性的调用,通常采用类.属性或者实例.属性的形式,举个小例子可能更容易理解python的属性。
class Test(object):
x = 10
def foo():
return Test.x >>> Test.x
10
>>> Test.foo()
10
通过Test.x及Test.foo()的方式可以类属性的方式调用,此处单例模式的实现也是一样,虽然使用类属性这个特性实现的单例和通过@staticmethod / @classmethod特性实现的单例从代码角度来说很相似,但是调用原理完全不同。
单线程实现
class Singleton(object):
_instance = None def __init__(self):
raise SyntaxError('can not instance, please use get_instance') def get_instance():
if Singleton._instance is None:
Singleton._instance = object.__new__(Singleton)
return Singleton._instance
6. 元类__metaclass__
原理分析
python中的metaclass比较复杂晦涩,不打算大篇幅介绍,这里主要应用元类的两点特性:
- type(name, bases, dict): 动态创建类。
- name: 类的名称
- bases: 基类的元组
- dict: 类内定义的namespace变量
- __metaclass__: 自定义元类
我们可以通过type定义一个元类,并把它返回的类(新建的元类)赋值给另一个类的__metaclass__属性,这样另一个类就具备了type创建的类的所有特性。
单线程实现
class Singleton(type):
def __init__(cls, name, bases, dict):
super(Singleton, cls).__init__(name, bases, dict)
cls._instance = None
def __call__(cls, *args, **kw):
if cls._instance is None:
cls._instance = super(Singleton, cls).__call__(*args, **kw) return cls._instance
class MySingletonClass(object):
__metaclass__ = Singleton # 或者使用如下格式
# class MySingletonClass(metaclass=Singleton):
# pass
7. 方法装饰器
原理分析
装饰器是基于AOP实现的,可以动态地改变类或函数的功能,具有很高的解耦性和灵活性,在python中应用非常广泛。
用函数装饰器实现单例模式的原理是:先创建一个可以传入类对象的外层函数,在该外层函数中创建一个instance字典来保存单例,同时创建一个内层函数get_instance(用来返回单例),在该内层函数判断instance字典中是否包含单例,如果没有就创建单例并以键值对的形式保存在instance字典中,然后通过get_instance()返回单例,外层函数的返回值为内层函数名。
使用函数装饰器时记住添加functools模块中的@wraps修复函数名及文档属性。
单线程实现
from functools import wraps def singleton(cls):
_instance = {}
@wraps(cls)
def get_instance(*args, **kw):
if cls not in _instance:
_instance[cls] = cls(*args, **kw)
return _instance[cls]
return get_instance @singleton
class MySingletonClass(object):
pass
8. 类装饰器
原理分析
装饰器本身就是对所修饰的类、对象、函数功能的扩展,相当于按需定制、重新封装。如果想让装饰器正常工作,必须在装饰器内部返回一个可调用的对象,可调用的对象可以是函数,也可以是类,所以装饰器不仅可以是函数,也可以是类。
类装饰器主要是通过类的构造函数__init__()传入一个函数或类对象,然后重载__call__()并且返回一个函数或类对象,来达到装饰器的目的。
此处使用类装饰器实现单例模式通过__init__()传入的是类对象,重载__call__()返回的也是类对象。将自定义的@Singleton装饰器附加到MySingletonClass类上就会调用单例模式生成MySingletonClass的唯一实例。
单线程实现
class Singleton(object):
def __init__(self, cls):
self._cls = cls
self._instance = {} def __call__(self, *args, **kwargs):
if self._cls not in self._instance:
self._instance[self._cls] = self._cls(*args, **kwargs)
return self._instance[self._cls] @Singleton
class MySingletonClass(object):
pass
9. import 模块
原理分析
分析python模块加载机制之前先回顾一下java加载机制对于static的处理,我们知道静态代码块、静态属性和静态方法只会在类首次加载的时候初始化一次,而且是全局性的、不会加载第二次,后续直接调用即可。
python模块的加载也非常类似,第一次import module时系统会执行模块代码生成.pyc文件,第二次import时就会直接加载.pyc文件,不会再次执行模块代码。所以,如果我们把需要单例化的函数和数据封装在一个模块文件中import,就可以获得这个单例对象了。从import module的机制来说,python的模块天然支持单例模式。
由import module方式生成的单例需要在代码中先指定单例,可作为简单使用,灵活性较差。
单线程实现
将如下代码保存为mysingleton.py,并执行from mysingleton import singleton导入单例。
class Singleton(object):
def test(self):
print ("for test") singleton = Singleton()
10. 共享属性
原理分析
本文开始就提到单例模式一个比较常用的场景就是提供全局访问点,可以存放与程序生命周期一致的全局共享资源,所以有时候我们并不在意是否真正实例唯一,更关心的是能否提供所有实例共享的状态或资源。
在python中每个对象都有自己的命名空间,空间内的变量存储在对象的__dict__中。类本身也是对象,所以类也有自己的__dict__,类的__dict__是由类的所有实例共享的,对于类中任一实例的属性的修改,所有的实例都会受到影响。
根据类的__dict__特性生成的实例并不唯一,实例id也会不同,严格地说并不属于单例模式的范畴,但此处更看重的是与单例模式一致的提供访问全局共享资源的功能。
单线程实现
class Borg(object):
_shared_state = {} def __init__(self):
self.__dict__ = self._shared_state class MySingletonClass(Borg):
def __init__(self, name):
super(MySingletonClass, self).__init__()
self.name = name def __str__(self):
return self.name >>> a = MySingletonClass("first")
>>> b = MySingletonClass("second")
>>> print (a, b)
second second
>>> print (id(a), id(b))
34292112 34197072
>>>