单例模式 (Singleton Pattern) 1. 它是怎么来的? 在软件开发中,我们经常会遇到一些“只需要一个,多了就会乱套”的资源。比如:
电脑的回收站(你不需要两个回收站,它们会打架)。
数据库连接池(频繁创建和销毁连接非常消耗性能,保持一个全局可复用的连接池更高效)。
应用程序的全局配置(如果每个模块都自己读一份配置,万一中途改了,数据就不一致了)。 在过去,程序员们通过全局变量来解决这个问题。但全局变量不仅不安全(容易被不小心覆盖),还破坏了面向对象的封装性。 于是,Gang of Four(四人帮,设计模式的奠基者们)提出了单例模式 :确保一个类只有一个实例,并提供一个全局访问点。
2. 适用场景
资源共享/控制:如日志记录器(Logger)、线程池、缓存。
统一配置管理:如读取 config.ini 或环境变量的类。
硬件接口访问:如打印机后台处理程序、显卡驱动接口。
3. 优缺点分析
优点
缺点
严格控制实例数目:节约系统资源,避免频繁创建/销毁的开销。
违反单一职责原则:它既要负责自身的业务逻辑,又要控制自己的实例化过程。
全局唯一访问点:避免了在不同模块之间传来传去。
不易测试:单例的全局状态在单元测试中很难被隔离或 Mock(模拟)。
延迟初始化(Lazy Initialization):只有在第一次用它时才创建,不占空闲内存。
多线程并发问题:在多线程环境下,如果不加锁,可能会意外创建出多个实例。
4. Python 代码实现 Python 有很多种实现单例的方法(比如模块导入、装饰器、__new__ 方法、元类)。这里我们展示最经典的两种。
💡 简单实例:利用 __new__ 方法 Python 中, __new__ 是真正创建实例的方法。我们可以拦截它,让它每次都返回同一个实例。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 class SimpleSingleton : _instance = None def __new__ (cls, *args, **kwargs ): if cls._instance is None : cls._instance = super ().__new__(cls) return cls._instance def __init__ (self, name ): self .name = name s1 = SimpleSingleton("Instance A" ) s2 = SimpleSingleton("Instance B" ) print (f"s1 的名字: {s1.name} " ) print (f"s2 的名字: {s2.name} " ) print (f"s1 和 s2 是同一个对象吗? {s1 is s2} " )
痛点:这个简单的写法在单线程下工作良好。但是在多线程高并发时,如果两个线程同时运行到 if cls._instance is None:,它们可能会同时创建实例,单例直接破产。
在 Python 中,最优雅、最工业级的单例实现方式是使用元类(Metaclass)并配合线程锁(Lock)。 元类是控制“如何创建类”的类,用它来实现单例,可以将单例逻辑和业务类完美剥离。 我们以一个数据库连接池为例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 import threadingimport timeclass SingletonMeta (type ): """ 一个线程安全的单例元类。 """ _instances = {} _lock: threading.Lock = threading.Lock() def __call__ (cls, *args, **kwargs ): if cls not in cls._instances: with cls._lock: if cls not in cls._instances: instance = super ().__call__(*args, **kwargs) cls._instances[cls] = instance return cls._instances[cls] class DatabaseConnectionPool (metaclass=SingletonMeta): """ 业务类:数据库连接池 """ def __init__ (self, db_url: str ): self .db_url = db_url print (f"[Init] 正在连接数据库 {self.db_url} ..." ) time.sleep(1 ) print ("[Init] 数据库连接池初始化成功!" ) def get_connection (self ): return f"Database connection from {self.db_url} " def worker (thread_id ): db_pool = DatabaseConnectionPool("mysql://localhost:3306/my_db" ) print (f"线程 {thread_id} 获取到的连接: {db_pool.get_connection()} (内存地址: {id (db_pool)} )" ) threads = [] for i in range (3 ): t = threading.Thread(target=worker, args=(i,)) threads.append(t) t.start() for t in threads: t.join()
运行结果分析:尽管 3 个线程几乎同时发起请求,但 [Init] 的初始化日志只打印了一遍,且所有线程拿到的对象内存地址(id)完全相同。这说明我们的单例在高并发下经受住了考验!