Python 使用__new__方法实现单例模式

__new__方法实现单例模式

class SingletomCls:

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls,"_instance"):
            cls._instance=super(SingletomCls, cls).__new__(cls)
        return cls._instance
     def __init__(self,*args,**kwargs):
        pass
 
 class Foo(SingletomCls):
    def __init__(self,name):
        self.name=name


ls=Foo("ls")
print(ls.name)

zs=Foo("zs")
print(ls.name)
print(zs.name)

代码打印结果：

ls
zs
zs

__new__实现的单例模式，可以实现重新初始化，因为控制单例的节点是在创建阶段，可以重新使用__init__进行初始化。

上面实现的单例，不是线程安全的。比如线程A，判断了没有实例，即not hasattr(cls,"_instance")为True，下一步让出CPU使用权给线程B。线程B，获得使用权，重新判断有无实例，无实例创建实例。然后A线程重新获得调度，继续去创建实例。造成系统中有多个实例存在。

写了个代码去验证这个线程不安全，为了夸大效果，在__new__里，有个等待时间。

import threading
import time
class SingletomCls:
    _single_lock=threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls,"_instance"):
            time.sleep(5)
            cls._instance=super(SingletomCls, cls).__new__(cls)
        return cls._instance

    def __init__(self,*args,**kwargs):
        pass


class Foo(SingletomCls):
    def __init__(self,name):
        self.name=name

def task(args):
    obj=Foo("name_"+str(args))
    print(obj)


for i in range(10):
    t=threading.Thread(target=task,args=(i,))
    t.start()

结果：

<__main__.Foo object at 0x00000246E6636448>
<__main__.Foo object at 0x00000246E663C388>
<__main__.Foo object at 0x00000246E663C5C8><__main__.Foo object at 0x00000246E663C388><__main__.Foo object at 0x00000246E663C308>
<__main__.Foo object at 0x00000246E663C3C8><__main__.Foo object at 0x00000246E663C308>
<__main__.Foo object at 0x00000246CDDAFC48>
<__main__.Foo object at 0x00000246CDDC1348><__main__.Foo object at 0x00000246CDDAFC48>

可以发现，在这种多线程情况下，不能保证单例！

__new__方法实现带锁的单例模式

import threading

class SingletomCls:
    _single_lock=threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls,"_instance"):
            with cls._single_lock:
                if not hasattr(cls,"_instance"):
                    cls._instance=super(SingletomCls, cls).__new__(cls)
        return cls._instance


    def __init__(self,*args,**kwargs):
        pass


class Foo(SingletomCls):
    def __init__(self,name):
        self.name=name

def task(args):
    obj=Foo("name_"+str(args))
    print(obj)

for i in range(10):
    t=threading.Thread(target=task,args=(i,))
    t.start()

打印的结果：

<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>
<__main__.Foo object at 0x00000223F4C61348>

多线程情况下，仍然只有一个实例存在

加入等待时间，验证效果：

import threading
import time
class SingletomCls:
    _single_lock=threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls,"_instance"):
            time.sleep(5)
            with cls._single_lock:
                if not hasattr(cls,"_instance"):
                    cls._instance=super(SingletomCls, cls).__new__(cls)
        return cls._instance
        
    def __init__(self,*args,**kwargs):
        pass


class Foo(SingletomCls):
    def __init__(self,name):
        self.name=name

def task(args):
    obj=Foo("name_"+str(args))
    print(obj)

for i in range(10):
    t=threading.Thread(target=task,args=(i,))
    t.start()

<__main__.Foo object at 0x0000021B672BFC48><__main__.Foo object at 0x0000021B672BFC48><__main__.Foo object at 0x0000021B672BFC48>
<__main__.Foo object at 0x0000021B672BFC48>

<__main__.Foo object at 0x0000021B672BFC48>

<__main__.Foo object at 0x0000021B672BFC48><__main__.Foo object at 0x0000021B672BFC48><__main__.Foo object at 0x0000021B672BFC48><__main__.Foo object at 0x0000021B672BFC48>
<__main__.Foo object at 0x0000021B672BFC48>