vlambda博客
学习文章列表

5分钟用Python创建一个单例模式

单例模式(Singleton Pattern),就是整个程序有且仅有一个相同的实例。该类负责创建自己的对象,同时确保只有一个对象被创建

应用场景

  1. 你希望这个类只有一个且只能有一个实例,如某个初始化含登录的用户类,你不想在多处实例同一个用户仍需登录多次;;
  2. 项目中的一些全局管理类(Manager)可以用单例来实现。

类似案例:Python logging 模块的 getLogger

# ~/Lib/logging/__init__.py
class Manager(object):
    def getLogger(self, name):
        """
        Get a logger with the specified name (channel name), creating it
        if it doesn't yet exist. This name is a dot-separated hierarchical
        name, such as "a", "a.b", "a.b.c" or similar.

        If a PlaceHolder existed for the specified name [i.e. the logger
        didn't exist but a child of it did], replace it with the created
        logger and fix up the parent/child references which pointed to the
        placeholder to now point to the logger.
        """

        rv = None
        if not isinstance(name, str):
            raise TypeError('A logger name must be a string')
        _acquireLock()
        try:
            if name in self.loggerDict:
                rv = self.loggerDict[name]
                if isinstance(rv, PlaceHolder):
                    ph = rv
                    rv = (self.loggerClass or _loggerClass)(name)
                    rv.manager = self
                    self.loggerDict[name] = rv
                    self._fixupChildren(ph, rv)
                    self._fixupParents(rv)
            else:
                rv = (self.loggerClass or _loggerClass)(name)
                rv.manager = self
                self.loggerDict[name] = rv
                self._fixupParents(rv)
        finally:
            _releaseLock()
        return rv

def getLogger(name=None):
    """
    Return a logger with the specified name, creating it if necessary.
    If no name is specified, return the root logger.
    """

    if name:
        return Logger.manager.getLogger(name)
    else:
        return root

不同的姿势实现单例

5分钟用Python创建一个单例模式

基于 __new__ 方式实现

不考虑线程安全的情况下实现单例模式

class SimpleSingleton:
    def __init__(self, fuid=10001):
        self.fuid = fuid

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            cls._instance = super().__new__(cls)
        return cls._instance
<__main__.SimpleSingleton object at 0x00000250F3133308>
<__main__.SimpleSingleton object at 0x00000250F3133308>
<__main__.SimpleSingleton object at 0x00000250F3133308>

考虑线程安全的单例模式

上面创建单例虽然简单,但如果在 __new__ 方法中有一些耗时操作,在使用多线程时很可能会变成非单例模式,原因很简单,多个线程都进入了实例生成阶段。

import time
import threading

class SimpleSingleton:
    def __init__(self, fuid=10001):
        self.fuid = fuid

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            time.sleep(1)  # 模拟耗时操作
            cls._instance = super().__new__(cls)
        return cls._instance

def task(arg):
    obj = SimpleSingleton()
    print(obj)

for i in range(3):
    t = threading.Thread(target=task,args=[i,])
    t.start()
<__main__.SimpleSingleton object at 0x0000020D45E85C88>
<__main__.SimpleSingleton object at 0x0000020D45E85288>
<__main__.SimpleSingleton object at 0x0000020D45E85908>

这时返回的就不是同一个实例了,应该如何避免呢?加锁

class Myclass:
    _instance_lock = threading.Lock()  # 支持多线程的单例模式

    def __init__(self, fuid=10001):
        self.fuid = fuid
        print('call __init__')

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            with cls._instance_lock:
                time.sleep(1)  # 模拟耗时操作
                if not hasattr(cls, "_instance"):
                    cls._instance = super().__new__(cls)
        return cls._instance

obj1 = Myclass(10001)
obj2 = Myclass(fuid=10001)
obj3 = Myclass(fuid=10002)
print(obj1)
print(obj2)
print(obj3)
call __init__
call __init__
call __init__
<__main__.Myclass object at 0x00000197B57F0EC8>
<__main__.Myclass object at 0x00000197B57F0EC8>
<__main__.Myclass object at 0x00000197B5AB0C88>

1. 两层 if 判断(双重检测机制)

细心的朋友可能关注到上面面为什么用了两层if 判断,可以试想,两个thead执行到是否有_instance属性这个判断thead1先通过,执行下一句with cls._instance_lock,那么便直接占有了锁,之后在thread11sleepthread2也通过了第一个if判断,而继续执行执行with cls._instance_lock语句,无法抢占锁,被阻塞,当thread1完成1秒的sleep后,并且通过第二个if,对cls._instance赋值,退出with context后,thread2才能继续执行,1 秒sleep之后再进行第二个if判断,此时不能通过了,因为thread1已经创建了一个instance那么只好退出with context,再执行return cls._instance,其实就是返回thread1创建的cls._instance

2. __init__ 仍调用了多次

还有个问题,虽然我们得到的类实例都是同一个(0x00000197B57F0EC8),但__init__ 仍被调用了多次,如果在初始化进行数据库连接等操作时,也会连接多次,这不是我们想要的, 为什么是这样?

__new__负责对象的创建,而 __init__负责对象的初始化;__new__是一个类方法,而 __init__是一个对象方法。

__new__是我们通过类名进行实例化对象时自动调用的,__init__是在每一次实例化对象之后调用的,__new__方法创建一个实例之后返回这个实例对象,并将其传递给 __init__方法的 self 参数,也就是说 __new__返回的对象在真正完成实例化前还是会调用 __init__

How To:一种方案就是在 __init__调用时也做一次判断, 如下

class Myclass:
    _instance_lock = threading.Lock()  # 支持多线程的单例模式
    _is_first_init = False

    def __init__(self, fuid=10001):
        if not self._is_first_init:
            self.fuid = fuid
            print('call __init__')
            self._is_first_init = True

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            with cls._instance_lock:
                time.sleep(1)  # 模拟耗时操作
                if not hasattr(cls, "_instance"):
                    cls._instance = super().__new__(cls)
        return cls._instance

显然这是不优雅的,推荐使用下方的 metaclass装饰器 实现方式

基于 metaclass 方式实现 [推荐]

class SingletonMetaClass(type):
    _instance_lock = threading.Lock()  # 支持多线程的单例模式
    _instance = {}

    def __call__(cls, *args, **kwargs):
        fuid = args[0if args else kwargs.get('fuid'or 10001
        if not cls._instance.get(fuid):
            with cls._instance_lock:
                if not cls._instance.get(fuid):
                    cls._instance[fuid] = super().__call__(*args, **kwargs)

        return cls._instance[fuid]

 class Myclass(metaclass=SingletonMetaClass):
    def __init__(self, fuid=10001):
        self.fuid = fuid
        print('call __init__')

基于装饰器方式实现 [推荐]

def singleton_wrapper(cls, *args, **kwargs):
    """定义一个单例装饰器"""
    instance = {}

    def wrapper(*args, **kwargs):
        if cls not in instance:
            instance[cls] = cls(*args, **kwargs)
        return instance[cls]

    return wrapper
    
 @singleton_wrapper
 class Myclass:
    def __init__(self, fuid=10001):
        self.fuid = fuid
        print('call __init__')

嘿!你还在看吗