经典创建型设计模式：单例模式与多例模式
了解单例模式的动机
-
单例模式的核心要义是什么？简单描述
一个类只能有唯一一个实例，并提供一个全局访问点供其他模块调用。
-
为什么会需要只能存在唯一实例的类？
避免资源重复创建、减少对象创建与销毁的开销，造成不必要的性能消耗（如内存消耗）。尤其针对一些只需要维护一份资源的场景（如项目全局配置）和实例化耗时长、消耗多的资源。
避免多人并发操作时的数据混乱（如多人同时写入一个日志文件）
理解单例模式的唯一性
- 单例模式要求一个类只能有一个实例，是指线程中只能有一个实例？还是进程中只能有一个实例？还是多个进程间只能有一个实例？
经典单例模式中实例的唯一性是“进程唯一”、“线程唯一”还是“集群唯一”取决于实现，一般来说是“进程唯一”，即一个进程内单例类只存在一个类实例，进程内多个线程访问到的实例是同一个。
-
如何实现进程唯一的单例？
-
可否实现线程唯一的单例？同一进程内多个线程间的实例不相同，但同一个线程内访问的实例是唯一的。
可以，可以通过key-value结构维护线程ID与实例的映射关系，确保同一个进程内每个线程针对单例类只有一个实例，且该实例与其他线程的实例不相同。
-
可否实现进程间唯一的单例？
可以，要点是把单例对象的存储与访问放在多进程可以共享的存储空间，比如共享的文件、数据库等，通过分布式锁/进程锁确保同一时刻只能有一个进程访问资源，只创建唯一一份类实例。
这种情况下，一般类实例的存储需要经过序列化，进程加载类实例时需要进行反序列化成对象方可使用。
实现单例模式(基于Python实现)
单例模式有哪些实现？
根据单例创建的时机可以简单地对单例模式进行分类：饿汉式单例模式和懒汉式单例模式。
饿汉式单例模式
其中饿汉式单例模式一般指程序初始化时就对单例类进行实例化。
假如通过Python实现饿汉单例模式，最简单的一种思路就是利用python模块的单例特性，将单例类及其实例化实现在一个单独的模块中，然后其他模块通过对单例模块进行调用获取单例类的实例。
# singleton_module.py
"""
将单例放到一个单独的模块
"""
class _SingletonModule:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
self.config = None
config = _SingletonModule()
# main.py
from singleton_module import config, _SingletonModule
a = config
b = config
c = _SingletonModule()
print(a is b)
print(c is a)
main.py的输出为：
True
True
除了将单例类及其实例化放到单独的模块中，也可以将单例类的实例化放在类加载的时机：即将单例实例作为类静态成员的初始值，一种实现如下：
# hungry_singleton
"""
饿汉单例模式
"""
class SingletonModule:
class _SingletonModule:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
self.config = None
_instance = _SingletonModule()
def __new__(cls, *args, **kwargs):
return cls._instance
a = SingletonModule()
b = SingletonModule()
print(a is b)
# Output: True
采用饿汉式实现，可以将耗时的实例初始化操作提前到程序启动时进行，而非在实际提供服务时才进行耗时的初始化。
如果是一个耗费资源（内存、显存）较多的实例化过程，提前实例化也有助于提前暴露可能存在的问题。
懒汉式单例模式
懒汉式单例模式与饿汉式单例模式的不同之处在于单例的实例化时机不同，懒汉式允许程序在使用单例时再进行实例化，也就是“延迟加载”。
下面以一个加载配置文件的功能演示懒汉式单例模式。
# lazy_loading/main.py
import json
import threading
from dataclasses import dataclass, asdict
@dataclass
class ConfigStruct:
lesson: str
name: str
description: str
type: str
@classmethod
def from_json(cls, file_path):
with open(file_path, "r") as fr:
config_dict = json.load(fr)
return cls(**config_dict)
class NonSingletonConfiguration:
_config = None
def __init__(self, config_dict):
print(f"{threading.current_thread().name} creating config object")
self._config = self._setup(config_dict)
self.id = id(self)
def __getattr__(self, item):
if self._config is None:
self._setup(None)
return self._config.get(item, None)
def _setup(self, config_dict: dict):
if config_dict is None:
config_dict = {}
return config_dict
class SingletonConfiguration:
_config = None
def __new__(cls, **kwargs):
if not cls._config:
cls._config = super().__new__(cls)
return cls._config
def __init__(self, **kwargs):
config_dict = kwargs.get("config_dict", None)
thread_name = kwargs.get("name", threading.current_thread().name)
print(f"{thread_name} init config object: {id(self)}")
if config_dict is None:
config_dict = {}
self._config = config_dict
def __getattr__(self, item):
return self._config.get(item, None)
def create_singleton_worker(config_dict, name="T"):
return SingletonConfiguration(config_dict=config_dict, name=name)
if __name__ == "__main__":
from pathlib import Path
config_dict = ConfigStruct.from_json(f"{Path(__file__).parent}/config.json")
print(config_dict)
print("懒汉单例模式-----")
object_list = []
for _ in range(10):
object_list.append(SingletonConfiguration(config_dict=asdict(config_dict)))
print(all(object_list[0] is obj for obj in object_list))
print(object_list[0].lesson)
print("非单例模式-----")
c = NonSingletonConfiguration(asdict(config_dict))
d = NonSingletonConfiguration(asdict(config_dict))
print(c is d)
print(c.lesson)
threads = []
for i in range(10):
t = threading.Thread(target=create_singleton_worker, args=(asdict(config_dict), f"T{i}"))
threads.append(t)
t.start()
for t in threads:
t.join()
执行结果如下：
【Tips：可以观察到无论是单个线程中多次获取SingletonConfiguration类的实例，还是在多个线程中获取，得到的类实例都是同一个（具有同一个内存地址）】
懒汉式单例模式的缺点在于，并发场景时可能存在线程不安全问题，如多线程创建类实例时碰巧在不同的内存空间创建了实例，因此实际实现时需要还需要通过控制资源访问。
下面对比非线程安全和线程安全的单例
# @Author: weirdgiser
# @Function: 对比线程安全的单例模式和线程不安全的单例模式
import threading
import time
thread_count = 200
create_count_safe = 0
create_count_unsafe = 0
def timer(func):
def wrapper(*args, **kwargs):
start = time.time()
func(*args, **kwargs)
print(f"cost time: {time.time()-start}")
return wrapper
class UnSafeSingletonClass:
_config = None
def __new__(cls, **kwargs):
global create_count_unsafe
if not cls._config:
print(f"{threading.current_thread().name} creating config object")
create_count_unsafe += 1
cls._config = super().__new__(cls)
return cls._config
def __init__(self):
# print(f"{threading.current_thread().name} init config object: {id(self)}")
pass
class SafeSingletonClass:
_config = None
_lock = threading.RLock()
def __new__(cls, **kwargs):
global create_count_safe
with cls._lock:
if not cls._config:
print(f"{threading.current_thread().name} creating config object")
create_count_safe += 1
cls._config = super().__new__(cls)
return cls._config
def __init__(self):
# print(f"{threading.current_thread().name} init config object: {id(self)}")
pass
@timer
def test_safe():
print("懒汉单例模式-----")
threads = []
for i in range(thread_count):
t = threading.Thread(target=SafeSingletonClass, name=f"T{i}")
threads.append(t)
t.start()
for t in threads:
t.join()
print(create_count_safe)
@timer
def test_unsafe():
print("懒汉单例模式-----")
threads = []
for i in range(thread_count):
t = threading.Thread(target=UnSafeSingletonClass, name=f"T{i}")
threads.append(t)
t.start()
for t in threads:
t.join()
print(create_count_unsafe)
if __name__ == "__main__":
test_safe()
test_unsafe()
可见，多线程场景下UnSafeSingletonClass类派生出的实例可能具有多个，而SafeSingleton类派生出的实例始终只有一个。当然，由于线程需要在创建实例前进行上锁或释放锁，SafeSingleton的实例化会更加耗费时间。
-
你能否发现上述SafeSingletonClass存在的问题？有没有办法优化上述SafeSingletonClass类的实现？
稍微思考一下不难发现，多线程场景下，如果某一个（可能是第一个）执行的线程成功创建了实例，后续的线程还需要尝试获取锁吗？其实是不需要的，如果实例已经创建成功，直接返回即可，而SafeSingletonClass却要求每个线程都要尝试获取锁再判断实例是否创建完成，显然这降低了并发能力，可以通过如下思路优化：当线程发现实例已经创建好时，直接返回实例对象即可，若实例未创建，则尝试获取锁并看是否需要创建实例，这种做法也叫做双重检测，实现如下：
# double_check/main.py class SafeDoubleCheckSingletonClass: _config = None _lock = threading.RLock() def __new__(cls, **kwargs): global create_count_doublecheck if not cls._config: with cls._lock: if not cls._config: print(f"{threading.current_thread().name} creating config object") create_count_doublecheck += 1 cls._config = super().__new__(cls) return cls._config
单例模式的分类总结
-
饿汉式
程序初始化、模块/类加载时即对类进行实例化，一般是线程安全的。
-
懒汉式
延迟加载，等到使用类时才进行实例化，但需要应对多线程竞争条件问题。
-
双重检测
既考虑一般懒汉式单例的线程安全问题，同时减少不必要的锁获取次数。
-
静态内部类
Java中可以通过静态内部类实现线程安全的懒汉式单例模式
了解多例模式
-
多例模式的核心要义是？与工厂模式有什么区别？
一个类可以创建多个对象，并且可以限定可创建对象的数量。工厂模式用于创建不同类的实例，多例模式针对同一个类创建多个实例。
-
多例模式的应用场景有哪些？
-
数据库连接池
-
线程池
-
负载均衡器
-
租户数据管理
-
设备控制器
-
-
如何实现多例模式？
实现多例模式
-
如何维护和限制一个类的实例数量？
维护一个类成员变量用于计数
-
如何维护实例？
通过字典、数组等容器维护实例。
一种示例实现如下：
# @Author: weirdgiser
# @Function: 多例模式
import threading
_MAX_INSTANCE_LIMIT = 10
class MultitonClass:
_instances = {}
_max_instances = _MAX_INSTANCE_LIMIT
_lock = threading.Lock()
def __new__(cls, *args, **kwargs):
key = kwargs.get("name", threading.current_thread().native_id)
if key not in cls._instances:
with cls._lock:
if key not in cls._instances:
if len(cls._instances) >= cls._max_instances:
raise Exception("Max instances limit reached")
cls._instances[key] = super().__new__(cls)
print(f"instance {key} created")
return cls._instances[key]
if __name__ == "__main__":
thread_count = 20
threads = []
for i in range(thread_count):
t = threading.Thread(target=MultitonClass, kwargs={"name": f"thread-{i}"})
threads.append(t)
t.start()
for t in threads:
t.join()
a1 = MultitonClass(name="thread-1")
a2 = MultitonClass(name="thread-1")
b1 = MultitonClass(name="thread-2")
print(a1 is a2, a1 is b1)
