1. 程式人生 > >深入單例模式

深入單例模式

class Singleton(object):

    def __init__(self):
        pass

    @classmethod
    def instance(cls, *args, **kwargs):
        if not hasattr(Singleton, "_instance"):
            Singleton._instance = Singleton(*args, **kwargs)
        return Singleton._instance

import threading

def task(arg):
    obj = Singleton.instance()
    print(obj)

for i in range(10):
    t = threading.Thread(target=task,args=[i,])
    t.start()
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>
<__main__.Singleton object at 0x000002B9D0BE0240>

一般情況,大家以為這樣就完成了單例模式,但是這樣當使用多執行緒時會存在問題

在__init__新增模擬在做io操作:

def __init__(self):
    import time
    time.sleep(1)

會出現:

<__main__.Singleton object at 0x0000016F611C18D0>
<__main__.Singleton object at 0x0000016F61040278>
<__main__.Singleton object at 0x0000016F611F5F28>
<__main__.Singleton object at 0x0000016F611F5E80>
<__main__.Singleton object at 0x0000016F61204C18>
<__main__.Singleton object at 0x0000016F611C19B0>
<__main__.Singleton object at 0x0000016F6120C198>
<__main__.Singleton object at 0x0000016F6121A6A0>
<__main__.Singleton object at 0x0000016F6121A668>
<__main__.Singleton object at 0x0000016F6120C278>

這是因為多執行緒的執行緒鎖的問題。

解決辦法:加鎖!未加鎖部分併發執行,加鎖部分序列執行,速度降低,但是保證了資料安全

import time
import threading
class Singleton(object):
    _instance_lock = threading.Lock()

    def __init__(self):
        time.sleep(1)

    @classmethod
    def instance(cls, *args, **kwargs):
        with Singleton._instance_lock:#加鎖
            if not hasattr(Singleton, "_instance"):
                Singleton._instance = Singleton(*args, **kwargs)
        return Singleton._instance


def task(arg):
    obj = Singleton.instance()
    print(obj)
for i in range(10):
    t = threading.Thread(target=task,args=[i,])
    t.start()
time.sleep(20)
obj = Singleton.instance()
print(obj)

解決問題

<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>
<__main__.Singleton object at 0x00000222EEB830F0>

現在有個問題需要優化,就是一開始開多執行緒建立10個單例,加鎖是為了資料安全,建立完後再過五秒,此時再建立物件時先判斷之前有沒有例項物件有的話直接呼叫之前創立的物件,不需要執行緒鎖,影響效率所以稍微改寫。

import time
import threading
class Singleton(object):
    _instance_lock = threading.Lock()

    def __init__(self):
        time.sleep(1)

    @classmethod
    def instance(cls, *args, **kwargs):
        if not hasattr(Singleton,"__instance"):
            with Singleton._instance_lock:#加鎖
                if not hasattr(Singleton, "_instance"):
                    Singleton._instance = Singleton(*args, **kwargs)
        return Singleton._instance


def task(arg):
    obj = Singleton.instance()
    print(obj)
for i in range(10):
    t = threading.Thread(target=task,args=[i,])
    t.start()
time.sleep(1)
print("wait 5 seconds")
time.sleep(5)
obj = Singleton.instance()
print(obj)
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
<__main__.Singleton object at 0x000001FEE03010F0>
wait 5 seconds
<__main__.Singleton object at 0x000001FEE03010F0>

這種方式實現的單例模式,使用時會有限制,以後例項化必須通過 obj = Singleton.instance() 

如果用 obj=Singleton() ,這種方式得到的不是單例

通過上面例子,我們可以知道,當我們實現單例時,為了保證執行緒安全需要在內部加入鎖

我們知道,當我們例項化一個物件時,是先執行了類的__new__方法(我們沒寫時,預設呼叫object.__new__),例項化物件;然後再執行類的__init__方法,對這個物件進行初始化,所有我們可以基於這個,實現單例模式

import time
import threading
class Singleton(object):
    _instance_lock = threading.Lock()

    def __init__(self):
        time.sleep(1)
    def __new__(cls, *args, **kwargs):
        if not hasattr(Singleton,"__instance"):
            with Singleton._instance_lock:#加鎖
                if not hasattr(Singleton, "_instance"):
                    Singleton._instance = super().__new__(cls)
        return Singleton._instance


def task(arg):
    obj = Singleton()
    print(obj)
for i in range(10):
    t = threading.Thread(target=task,args=[i,])
    t.start()
time.sleep(2)
print("wait 5 seconds")
time.sleep(5)
obj = Singleton()
print(obj)
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
<__main__.Singleton object at 0x0000018A229F2160>
wait 5 seconds
<__main__.Singleton object at 0x0000018A229F2160>