Python线程安全单例模式实现方法
·
1. 使用模块导入(最简单)
Python模块本身就是天然的单例,因为模块在第一次导入时会生成.pyc文件,再次导入时直接加载。
# singleton.py
class Singleton:
def __init__(self):
self.data = "I'm a singleton"
def do_something(self):
print("Doing something")
_instance = Singleton()
def get_instance():
return _instance
# 在其他文件中使用
from singleton import get_instance
singleton = get_instance()
2. 使用装饰器 + 线程锁
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):
self.value = 0
def increment(self):
self.value += 1
# 测试
def test_singleton():
instance = ThreadSafeSingleton()
instance.increment()
print(f"Value: {instance.value}")
threads = [threading.Thread(target=test_singleton) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
3. 使用元类(Metaclass)
import threading
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 DatabaseConnection(metaclass=SingletonMeta):
def __init__(self, connection_string="default"):
self.connection_string = connection_string
print(f"Creating database connection: {connection_string}")
def query(self, sql):
return f"Executing: {sql}"
# 使用
db1 = DatabaseConnection("mysql://localhost:3306")
db2 = DatabaseConnection("postgres://localhost:5432")
print(db1 is db2) # True
print(db1.connection_string) # mysql://localhost:3306
4. 使用 __new__ 方法
import threading
class SingletonClass:
_instance = None
_lock = threading.Lock()
def __new__(cls, *args, **kwargs):
if cls._instance is None:
with cls._lock:
if cls._instance is None: # 双重检查锁定
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, value=0):
# 注意:__init__可能被多次调用
if not hasattr(self, '_initialized'):
self.value = value
self._initialized = True
# 使用
obj1 = SingletonClass(10)
obj2 = SingletonClass(20)
print(obj1 is obj2) # True
print(obj1.value) # 10 (不会被重新初始化为20)
5. 使用 threading.local(线程隔离的单例)
import threading
class ThreadLocalSingleton:
"""每个线程拥有自己的单例实例"""
_local = threading.local()
def __new__(cls, *args, **kwargs):
if not hasattr(cls._local, "instance"):
cls._local.instance = super().__new__(cls)
return cls._local.instance
def __init__(self, thread_id=None):
if not hasattr(self, '_initialized'):
self.thread_id = thread_id or threading.get_ident()
self._initialized = True
# 每个线程有自己的实例
def worker(thread_id):
instance = ThreadLocalSingleton(thread_id)
print(f"Thread {thread_id}: {id(instance)}")
threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
t.start()
for t in threads:
t.join()
6. 使用 atexit 清理资源
import threading
import atexit
class SingletonWithCleanup:
_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)
atexit.register(cls._cleanup)
return cls._instance
def __init__(self):
if not hasattr(self, '_initialized'):
self.resources = []
self._initialized = True
@classmethod
def _cleanup(cls):
"""程序退出时清理资源"""
if cls._instance:
print("Cleaning up resources...")
cls._instance.resources.clear()
def add_resource(self, resource):
self.resources.append(resource)
# 使用
singleton = SingletonWithCleanup()
singleton.add_resource("Resource1")
推荐方案
对于大多数场景,推荐使用装饰器方式或元类方式:
# 推荐:装饰器方式(简单清晰)
@singleton
class ConfigManager:
def __init__(self):
self.config = {}
self.load_config()
def load_config(self):
# 加载配置
pass
# 或元类方式(更Pythonic)
class Logger(metaclass=SingletonMeta):
def __init__(self):
self.logs = []
def log(self, message):
self.logs.append(message)
注意事项
- 双重检查锁定:在多线程环境中,使用双重检查锁定避免不必要的锁竞争
__init__多次调用问题:使用标志位避免重复初始化- 子类问题:单例模式可能不适用于需要继承的场景
- 序列化/反序列化:注意反序列化时可能创建新实例
- 模块级单例:最简单且线程安全,但不够灵活
更多推荐
所有评论(0)