文本是《设计模式(共12篇)》专题的第 4 篇。阅读本文前,建议先阅读前面的文章:
单例模式详解 - Python版本
1️⃣ 概念
定义:保证一个类仅有一个实例,并提供一个全局访问点
类型:创建型
2️⃣ 适用场景
想确保任何情况下都绝对只有一个实例
3️⃣ 优点
- 在内存中只有一个实例,减少了内存开销
 - 可以避免对资源的多重占用
 - 设置了全局访问点,严格控制访问
 
4️⃣ 缺点
- 没有接口,扩展困难
 
5️⃣ 重点
- 私有构造器
 - 线程安全
 - 延迟加载
 - 序列化和反序列化安全
 - 反射
 
6️⃣ 单例模式Coding(懒汉式)
① 懒汉式简单版实现
class LazySingletonV1:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
问题:这个版本的实现会在多线程环境中出现问题。可以看到通过使用Python的threading模块模拟多线程的情况,我们拿到了两个不一样的对象。
② 实现线程安全的懒汉式
import threading
class LazySingletonV2:
    _instance = None
    _lock = threading.Lock()
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance
分析:通过使用threading.Lock()同步锁来实现懒汉式单例的线程安全是一种较为普遍的解决方案,但是此方案也有一定的局限。同步锁有上锁和解锁的开销所以此解决方案会存在性能开销过大的问题。
③ DoubleCheck双重检查实现懒汉式
import threading
class LazyDoubleCheckSingleton:
    _instance = None
    _lock = threading.Lock()
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance
分析:虽然这种方式兼顾了性能和安全同时也满足懒加载的情况,但是这种情况也有一定的缺陷。首先通过threading.Lock()我们保证了多线程情况下只有一个线程可以创建对象,如果对象已经被创建则直接返回不需要再进行加锁的操作,避免了性能的开销。
问题:在多线程环境下,由于线程0并没有初始化完成对象,但是线程1已经将此对象判断为非空,也就是说线程1拿到的其实是线程0正在进行初始化的对象,在这样的情况下系统就会报异常了。
④ 通过装饰器实现线程安全的单例
import threading
from functools import wraps
def singleton(cls):
    instances = {}
    lock = threading.Lock()
    @wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            with lock:
                if cls not in instances:
                    instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance
@singleton
class ThreadSafeSingleton:
    def __init__(self):
        pass
分析:通过装饰器模式,我们可以优雅地实现线程安全的单例模式,同时解决了多线程情况下的线程安全问题。
⑤ 通过模块实现单例
# singleton_module.py
class _SingletonClass:
    def __init__(self):
        self.data = None
    def set_data(self, data):
        self.data = data
    def get_data(self):
        return self.data
# 模块级别的实例
instance = _SingletonClass()
def get_instance():
    return instance
分析:这种解决方案实际上是基于Python模块的导入机制,Python模块在首次导入时会被执行一次,之后的导入都会返回缓存的模块对象。
7️⃣ 单例模式Coding(饿汉式)
① 饿汉式简单实现
class HungrySingleton:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    @classmethod
    def get_instance(cls):
        return cls()
# 类加载时即创建
_hungry_instance = HungrySingleton()
优点:模块加载时即创建,避免了线程安全问题
缺点:可能会导致内存的浪费
② 序列化破坏单例模式原理解析及解决方案
问题演示:
import pickle
class HungrySingleton:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
# 测试代码
def test_serialization():
    instance = HungrySingleton.get_instance()
    # 序列化
    with open('singleton_file.pkl', 'wb') as f:
        pickle.dump(instance, f)
    # 反序列化
    with open('singleton_file.pkl', 'rb') as f:
        new_instance = pickle.load(f)
    print(instance)
    print(new_instance)
    print(instance is new_instance)
从上边的测试中,我们可以看出通过序列化和反序列化我们得到了两个不同的对象,这样就违背了单例的初衷。
解决方案:
import pickle
class HungrySingleton:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
    def __reduce__(self):
        return (self.__class__.get_instance, ())
原理分析:针对这个问题,我们需要重写__reduce__方法。通过重写__reduce__方法,我们可以控制对象的序列化和反序列化过程,确保反序列化时返回的是同一个实例对象。
③ 单例模式反射攻击的解决方案
问题演示:
def test_reflection():
    cls = HungrySingleton
    instance = HungrySingleton.get_instance()
    new_instance = cls.__new__(cls)
    print(instance)
    print(new_instance)
    print(instance is new_instance)
我们可以看到通过直接调用__new__方法我们依然可以得到两个对象,那我们该怎么解决这样的问题呢?
解决方案:
class HungrySingleton:
    _instance = None
    _initialized = False
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    def __init__(self):
        if HungrySingleton._initialized:
            raise RuntimeError("单例构造器禁止重复调用")
        HungrySingleton._initialized = True
    @classmethod
    def get_instance(cls):
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
    def __reduce__(self):
        return (self.__class__.get_instance, ())
这样就可以解决这个问题了,但是这种解决方案只适用于非延时加载的单例模式,如果是延时加载的单例模式我们依旧可以通过直接调用来创建,那么有没有既能保证单例不被序列化破坏又能禁止反射创建的单例模式呢?
8️⃣ 单例模式的其他实现
① 使用new方法实现单例
class SingletonMeta(type):
    _instances = {}
    _lock = threading.Lock()
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            with cls._lock:
                if cls not in cls._instances:
                    cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]
class EnumInstance(metaclass=SingletonMeta):
    def __init__(self):
        self.data = None
    def print_test(self):
        print("Enum Print Test")
    def get_data(self):
        return self.data
    def set_data(self, data):
        self.data = data
    @classmethod
    def get_instance(cls):
        return cls()
序列化验证:
import pickle
def test_serialization():
    instance = EnumInstance.get_instance()
    instance.set_data(object())
    # 序列化
    with open('singleton_file.pkl', 'wb') as f:
        pickle.dump(instance, f)
    # 反序列化
    with open('singleton_file.pkl', 'rb') as f:
        new_instance = pickle.load(f)
    print(instance.get_data())
    print(new_instance.get_data())
    print(instance.get_data() is new_instance.get_data())
结论:可以看到元类单例可以很好地解决上述的问题。
② 基于容器的单例模式
class ContainerSingleton:
    def __init__(self):
        pass
    _singleton_map = {}
    @classmethod
    def put_instance(cls, key, instance):
        if key and instance is not None:
            if key not in cls._singleton_map:
                cls._singleton_map[key] = instance
    @classmethod
    def get_instance(cls, key):
        return cls._singleton_map.get(key)
注意:容器单例模式并不是线程安全的,不过这种单例模式也是有应用场景的,当一个程序中单例比较多时,可以使用这样的模式进行统一管理。
③ ThreadLocal线程单例
这种单例并不能保证全局唯一,但是可以保证线程唯一。
import threading
class ThreadLocalInstance:
    _local = threading.local()
    def __new__(cls):
        if not hasattr(cls._local, 'instance'):
            cls._local.instance = super().__new__(cls)
        return cls._local.instance
    @classmethod
    def get_instance(cls):
        return cls()
测试代码:
import threading
class TestRunnable:
    def run(self):
        instance = ThreadLocalInstance.get_instance()
        print(f"{threading.current_thread().name}  {instance}")
def test_thread_local():
    print(f"main thread {ThreadLocalInstance.get_instance()}")
    print(f"main thread {ThreadLocalInstance.get_instance()}")
    print(f"main thread {ThreadLocalInstance.get_instance()}")
    def thread_func():
        runner = TestRunnable()
        runner.run()
    t1 = threading.Thread(target=thread_func, name="Thread-1")
    t2 = threading.Thread(target=thread_func, name="Thread-2")
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("End")
9️⃣ 单例模式的应用
① 单例模式在Python标准库中的应用
# 类似于Java的Runtime类,Python中的sys模块就是一个单例
import sys
class SystemInfo:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    def get_version(self):
        return sys.version
    def get_platform(self):
        return sys.platform
    @classmethod
    def get_instance(cls):
        """
        Returns the system info object associated with the current Python application.
        Returns:
            the SystemInfo object associated with the current Python application.
        """
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance
# 在模块加载时创建实例
_current_system_info = SystemInfo()
def get_system_info():
    return _current_system_info
                    