Python中如何实现线程安全的单例模式?

Number of views 41

在多线程环境下,传统的单例模式可能创建多个实例,需要确保线程安全且高效的实现方式

1 Answers

在Python中实现线程安全的单例模式有几种方法。这里介绍两种常见的方法:使用__new__方法和使用装饰器。

方法1: 使用 __new__ 方法

import threading

class Singleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super(Singleton, cls).__new__(cls, *args, **kwargs)
        return cls._instance

这种方法通过在__new__方法中添加一个锁来确保线程安全。当第一次创建实例时,会先检查_instance是否为None,如果为None,则获取锁,并再次检查_instance是否为None(以避免竞争条件),然后创建实例。

方法2: 使用装饰器

装饰器是一种更简洁的方法来实现线程安全的单例模式。

import threading

def singleton(cls):
    instances = {}
    lock = threading.Lock()

    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 MySingletonClass:
    def __init__(self):
        self.value = "Singleton Instance"

# 测试代码
if __name__ == "__main__":
    import threading

    def test_singleton():
        instance = MySingletonClass()
        print(f"Thread {threading.current_thread().name}: {id(instance)}")

    threads = []
    for i in range(5):
        thread = threading.Thread(target=test_singleton, name=f"Thread-{i+1}")
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

在这个例子中,我们定义了一个名为singleton的装饰器,它使用一个字典instances来存储类的实例,并使用一个锁来确保线程安全。当我们尝试创建一个类的新实例时,装饰器会检查该类是否已经存在于字典中,如果没有,则获取锁并创建实例。

这两种方法都可以确保在多线程环境下只有一个实例被创建,从而实现线程安全的单例模式。选择哪种方法取决于你的具体需求和个人偏好。