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)

注意事项

  1. 双重检查锁定:在多线程环境中,使用双重检查锁定避免不必要的锁竞争
  2. __init__多次调用问题:使用标志位避免重复初始化
  3. 子类问题:单例模式可能不适用于需要继承的场景
  4. 序列化/反序列化:注意反序列化时可能创建新实例
  5. 模块级单例:最简单且线程安全,但不够灵活
Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐