单例模式

单例模式（Singleton Pattern）是最简单也最常用的设计模式之一，它确保一个类只能创建一个实例，并提供全局访问点。

问题定义

场景1：数据库连接池

# ❌ 问题：每次都创建新连接
class Database:
def __init__(self):
self.connection = self._connect()
def _connect(self):
print("创建新的数据库连接...")
return "DB_CONNECTION"
# 使用
db1 = Database()  # 创建连接1
db2 = Database()  # 创建连接2 - 浪费资源！

问题： - 多个数据库连接浪费资源 - 可能超过数据库的最大连接数 - 状态不一致的风险

场景2：配置管理器

# ❌ 问题：配置分散，可能不一致
class Config:
def __init__(self):
self.settings = {"debug": True, "port": 8000}
# 使用
config1 = Config()
config1.settings["port"] = 9000
config2 = Config()  # 新实例，port还是8000！
print(config2.settings["port"])  # 8000 - 不一致！

解决方案

单例模式确保一个类只有一个实例，并提供全局访问点。

标准实现

方法1：使用 __new__

class Singleton:
"""单例模式 - 使用__new__"""
_instance = None  # 类变量存储唯一实例
def __new__(cls, *args, **kwargs):
"""控制实例化过程"""
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
"""初始化（只执行一次）"""
if not hasattr(self, '_initialized'):
self.data = []
self._initialized = True
# 验证
s1 = Singleton()
s2 = Singleton()
print(s1 is s2)  # True，同一个实例
print(id(s1))    # 14023456789
print(id(s2))    # 14023456789 - 相同！
s1.data.append(1)
print(s2.data)   # [1] - 共享数据

方法2：使用装饰器

def singleton(cls):
"""单例装饰器"""
instances = {}
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
class Database:
"""数据库类"""
def __init__(self):
print("初始化数据库连接...")
self.connection = "DB_CONNECTION"
# 验证
db1 = Database()  # 初始化数据库连接...
db2 = Database()  # 不会重新初始化
print(db1 is db2)  # True

方法3：使用模块（Pythonic方式）

# singleton_module.py
class _Database:
def __init__(self):
self.connection = "DB_CONNECTION"
def query(self, sql):
print(f"执行: {sql}")
# 创建单例实例
database = _Database()

# 使用时
from singleton_module import database
database.query("SELECT * FROM users")  # 永远是同一个实例

线程安全的单例

问题：多线程环境

import threading
import time
class UnsafeSingleton:
_instance = None
def __new__(cls):
if cls._instance is None:
time.sleep(0.1)  # 模拟延迟
cls._instance = super().__new__(cls)
return cls._instance
# 多线程测试
def create_singleton():
instance = UnsafeSingleton()
print(f"实例ID: {id(instance)}")
threads = []
for _ in range(5):
t = threading.Thread(target=create_singleton)
threads.append(t)
t.start()
for t in threads:
t.join()
# 可能输出5个不同的ID！

解决方案1：加锁

import threading
class ThreadSafeSingleton:
_instance = None
_lock = threading.Lock()  # 类锁
def __new__(cls):
# 双重检查锁定（Double-Checked Locking）
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance

解决方案2：使用类变量初始化

class ThreadSafeSingleton:
_instance = None
# 在类加载时初始化（Python保证线程安全）
def __init__(self):
if ThreadSafeSingleton._instance is None:
self.data = []
ThreadSafeSingleton._instance = self
# 获取实例
def get_instance():
if ThreadSafeSingleton._instance is None:
return ThreadSafeSingleton()
return ThreadSafeSingleton._instance

实战应用

应用1：数据库连接池

import threading
class DatabasePool:
"""数据库连接池 - 单例模式"""
_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
def __init__(self):
if not hasattr(self, 'initialized'):
self.connections = []
self.max_connections = 10
self._initialize_pool()
self.initialized = True
def _initialize_pool(self):
"""初始化连接池"""
for i in range(self.max_connections):
self.connections.append(f"Connection-{i}")
print(f"初始化连接池，共{self.max_connections}个连接")
def get_connection(self):
"""获取连接"""
if self.connections:
return self.connections.pop()
else:
raise Exception("连接池已满")
def release_connection(self, conn):
"""释放连接"""
self.connections.append(conn)
# 使用
pool1 = DatabasePool()  # 初始化连接池
pool2 = DatabasePool()  # 返回同一个实例
conn = pool1.get_connection()
print(f"获取连接: {conn}")
pool2.release_connection(conn)  # 同一个实例，可以操作

应用2：配置管理器

class ConfigManager:
"""配置管理器 - 单例模式"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if not hasattr(self, 'loaded'):
self.config = {}
self._load_config()
self.loaded = True
def _load_config(self):
"""加载配置（从文件或环境变量）"""
self.config = {
"debug": True,
"database": {
"host": "localhost",
"port": 5432
},
"api": {
"key": "secret",
"timeout": 30
}
}
print("配置已加载")
def get(self, key, default=None):
"""获取配置"""
keys = key.split('.')
value = self.config
for k in keys:
value = value.get(k, default)
if value is None:
return default
return value
def set(self, key, value):
"""设置配置"""
self.config[key] = value
# 使用
config1 = ConfigManager()
config2 = ConfigManager()
print(config1 is config2)  # True
print(config1.get("database.host"))  # localhost
config2.set("debug", False)  # 同一个实例
print(config1.get("debug"))  # False

应用3：日志记录器

import logging
from datetime import datetime
class Logger:
"""日志记录器 - 单例模式"""
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
if not hasattr(self, 'initialized'):
self.logger = logging.getLogger('app')
self.logger.setLevel(logging.INFO)
self._setup_handler()
self.initialized = True
def _setup_handler(self):
"""设置日志处理器"""
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def info(self, message):
"""记录信息"""
self.logger.info(message)
def error(self, message):
"""记录错误"""
self.logger.error(message)
def warning(self, message):
"""记录警告"""
self.logger.warning(message)
# 使用
logger1 = Logger()
logger2 = Logger()
logger1.info("应用启动")
logger2.error("发生错误")  # 同一个logger

优缺点

✅ 优点

❌ 缺点

不适合的场景

❌ 需要多个实例的场景

# 单例不适合的场景：用户对象
class User(Singleton):  # ❌ 错误！
def __init__(self, name):
self.name = name
user1 = User("张三")
user2 = User("李四")
print(user2.name)  # 输出"张三"而不是"李四"！

正确做法：用普通类

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

❌ 可以用静态方法的场景

# 不需要单例，用静态方法即可
class MathUtils:
@staticmethod
def add(x, y):
return x + y
# 使用
result = MathUtils.add(1, 2)  # 无需实例化

单例模式的替代方案

1. 依赖注入

class Database:
def __init__(self, connection_string):
self.connection = connection_string
class UserService:
def __init__(self, database):
self.database = database
def get_user(self, user_id):
return self.database.query(user_id)
# 使用
db = Database("postgres://...")
user_service = UserService(db)
# 测试时可以注入Mock对象
test_db = MockDatabase()
test_service = UserService(test_db)

2. 模块级变量

# config.py
CONFIG = {
"debug": True,
"port": 8000
}
# 使用
from config import CONFIG
print(CONFIG["debug"])

本章要点

• ✅ 单例模式确保一个类只有一个实例
• ✅ 实现方式：__new__、装饰器、模块
• ✅ 多线程环境需要加锁保证线程安全
• ✅ 适用场景：数据库连接池、配置管理器、日志记录器
• ✅ 不适合需要多个实例的场景
• ✅ 可以用静态方法时无需单例
• ✅ 依赖注入是更好的替代方案

下一步：工厂方法模式 🚀