Python 创建线程安全的单例

之前在文章《深入理解 Python 中的类与元类》中讲了使用 __new__() 方法可以很方便地实现单例模式。但其实那段代码是线程不安全的。

验证代码如下：

import threading
import time

class Singleton(object):
    _instance = None
    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            time.sleep(1)
            cls._instance = object.__new__(cls)  
        
        return cls._instance

def create():
    obj = Singleton()
    print(obj)

threads = []
for i in range(10):
    t = threading.Thread(target=create)
    threads.append(t)

for t in threads:
    t.start()

输出结果如下图：

正确的线程安全的实现方法是：

import logging, threading

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

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = object.__new__(cls)
        
        return cls._instance

    def __init__(self):
        cls = self.__class__
        if not cls._initialized:
            with cls._lock:
                if not cls._initialized:
                    cls._initialized = True

                    # 在此处进行成员变量的声明与初始化
                    self.logger = logging.getLogger(cls.__name__)
                    self.logger.debug('Init %s singleton', cls.__name__)
        pass

为什么要嵌套两次判断 cls._instance is None 呢？

其实不加第一层判断也是可以的，但由于获取线程锁是一个“费时”的操作，所以加第一层判断其实就是为了避免每次调用 __new__() 方法都要去获取线程锁。

with lock: 的用法，参考官方文档。