Python設計模式 - 創建型 - 單例模式(Singleton) - 十種
對於很多開發人員來說,單例模式算是比較簡單常用、也是最早接觸的設計模式了,仔細研究起來單例模式似乎又不想看起來那麽簡單。我們知道單例模式適用於提供全局唯一訪問點,頻繁需要創建及銷毀對象等場合,的確方便了項目開發,但是單例模式本身也有一定的局限性,如果濫用則會給後續軟件框架的擴展和維護帶來隱患。
單例模式的實現有很多種,應用場合也各有不同,但必須保證實例唯一,如果是多線程環境則必須保證線程安全。python本身有很多內置特性可以用來實現單例模式的效果,理清每種單例模式的實現原理、優缺點及使用場合才能更好地用其長避其短。
定義
單例模式: 保證一個類只有一個實例,並提供一個訪問該實例的全局訪問點
應用場景
單例模式適用於很多場合,各個產品也有不少單例模式的實現例子,比如windows的回收站,網站計數器等。此處僅列舉跟軟件應用開發強相關的常用場景:
- 用於保存全局狀態,如全局的配置文件
- 用於避免頻繁創建及銷毀對象,如數據庫連接池
- 用於多線程之間共享對象資源,如線程池
優缺點
我們知道硬幣都有兩面,任何一種設計模式既有它的優勢,也會有它的局限性。所以非常有必要清楚每種設計模式的優缺點揚長避短。
- 優點
- 提供了全局訪問點,方便統一管理
- 避免了對同一資源的訪問競爭及操作沖突
- 減小了開銷,尤其是需要頻繁創建和銷毀的對象
- 類可以根據需要方便靈活地控制實例化過程
- 缺點
- 不適用於變化的對象,不可以根據不同的應用場景創建不同的對象
- 單例模式沒有抽象層,不易擴展及解耦
- 承擔的職責過多,違反單一職責原則
- 進程被殺時容易引發內存泄漏、及資源未釋放的問題
- 單例模式中創建新對象的方式和項目開發中默認的創建新對象的方式不一致,這會隱性增加開發成本
單例模式的校驗
實例唯一
單例模式的出發點就是通過唯一的實例提供全局訪問點,當然在多線程環境下還有線程安全的要求。這裏先說一下如何檢驗單例模式生成的實例是否唯一,線程安全的話題在下面。
python中的每個對象都包含三個要素:
- id: 唯一標識一個對象
- type: 標識對象的類型
- value: 對象的值
a is b: 用來判斷對象a是否就是對象b,是通過id來判斷的
a == b: 用來判斷對象a的值是否和對象b的值相等,是通過value來判斷的
所以如果運行中的單例模式每次都生成ID相同的實例,也就驗證了該單例模式的實例唯一性。
線程安全
線程安全簡單地說就是多個線程同時執行某個程序時不會發生寫沖突。
要想使程序在多線程環境下依舊運行正確,就必須保證同一時刻只有一個線程對共享數據進行存取,多線程環境下通常使用加鎖來保證存取操作的唯一性和排他性。線程安全一直是並發領域的一個難點,不管對於靜態語言還是動態語言都有很多內置的、常用的collections是非線程安全的,比如Java的HashMap本身是非線程安全的,使用不當的話會造成嚴重的性能問題,但是由於其應用非常廣泛,不少人對它的非線程安全性並不敏感,所以在多線程環境中一定要使用其對應的線程安全版本;再比如python內置的list,dict,set,iterator都不是線程安全的,需要我們自己加鎖進行線程同步。
就單例模式的線程安全來說,有兩個認識上的誤區,一是有人覺得單例模式本身只能生成唯一實例,ID都是同一個,既然是獨苗應該不會再有線程安全的問題,二是python本身提供的GIL機制本身就是保證同一時刻只有一個線程對共享數據進行存取的,所以python內置的GIL機制已經從原子操作方面保證線程安全了。事實真的是這樣嗎?這就需要先進行原理分析、再進行數據測試來檢驗了。
誤區一:實例唯一與線程安全
實例唯一性和線程安全性是兩個不同的範疇,實例唯一是保證只有一個全訪問點,這在單線程中當然是沒有問題的,但是在多線程環境中全局訪問點作為共享資源本身就是資源競爭的對象,有多個線程都想獲得這個唯一實例的操作權限。
我們知道軟件項目中有很多實現都是面向事務的,事務本身是由多個任務組合在一起的,要想保證事務的安全性就要求操作成功時提交整個事務,操作失敗時回滾整個事務,對於單例模式的唯一實例也是如此,試想如果線程A在某個時間片段獲得該唯一實例的操作權限,執行到一半任務時,因為時間片到、中斷、函數調用等原因交出操作權限,線程B獲得該唯一實例的控制權,此時如果線程A和線程B之間相互獨立、沒有共享存取資源的話當然沒問題,如果有共享資源則很容易出現數據或事務汙染,也就是非線程安全,無法保證程序正確運行。所以即使是唯一實例也需要加鎖保證對線程安全敏感的事務能夠正確提交或整體回滾。
誤區二:GIL與原子操作
GIL的官方解釋是:全局解釋器鎖。每個線程在執行到過程中都需要先獲取GIL,保證同一時刻只有一個線程可以執行代碼。
這聽起來很線程安全,但是問題是GIL的線程安全是基於原子操作的,而我們實際的執行語句或執行代碼是由一個或多個原子操作組成的,如果每條語句或者每塊代碼整體上都是一個原子操作這當然也沒有問題,但事實上這是不可能的,有些語句比如x = x + 1本身就表示tmp = x + 1及x = tmp(tmp表示中間變量)兩個原子操作,也就是說這條語句在GIL機制下無法保證線程安全。如果在多線程下執行 x = x + 1這條語句,不對其加鎖進行同步的話無法保證最終計算結果的正確性。
非線程安全的單例實現
上面說的都是理論上的解釋,最好有實際測試數據佐證看起來更清楚、直白,更有說服力。先看一下未加鎖的單例實現及其在多線程下的執行結果,因為單純的單例實現無法測試出該單例模式是否是線程安全的,所以我們給該單例模式加上相應的數據計算,如果數據計算最終結果與預期一致就表示線程安全,如果不一致就表示多線程環境下程序未能正常執行,該單例實現是非線程安全的,當然為了避免偶然性的發生,我們可以多次執行同一個測試程序。
要添加的數據計算是
zero += 1
zero -= 1
從理論上說,先+1再-1,無論執行多少次,zero變量的值都應該是0。將這兩個語句封裝到單例中,並加大循環次數(數據量大更容易出現線程安全性問題,程序執行時間可能會多花幾秒種),保存成mysingleton.py:
class Singleton(object): def __init__(self): self.zero = 0 def __new__(cls, *args, **kw): if not hasattr(cls, "_instance"): cls._instance = super(Singleton, cls).__new__(cls, *args, **kw) return cls._instance def change_zero(self): for i in range(10000000): self.zero += 1 self.zero -= 1
封裝校驗類先檢驗但是是否唯一,再檢驗是否線程安全,並多次執行避免偶然性:
from mysingleton import Singleton
import threading
class SingletonCheck(object): def __init__(self): self.task = Singleton() def action(self): obj = Singleton() print ("Created Object: {}".format(obj)) def only_instance_test(self): for i in range(10): t = threading.Thread(target=self.action) t.start() t.join() def thread_safety_test(self): t1 = threading.Thread(target=self.task.change_zero) t2 = threading.Thread(target=self.task.change_zero) t3 = threading.Thread(target=self.task.change_zero) t1.start() t2.start() t3.start() t1.join() t2.join() t3.join() print ("期望的zero值: 0, 實際的zero值: {}\n該實例的實現是否是線程安全的: {}".format(self.task.zero, 0 == self.task.zero)) if __name__ == "__main__": singleton_check = SingletonCheck() print ("開始驗證該單例模式生成的實例ID是否唯一...") singleton_check.only_instance_test() print ("結束驗證該單例模式生成的實例ID是否唯一...") print ("=" * 64) print ("開始驗證該單例模式是否線程安全...") singleton_check.thread_safety_test() print ("結束驗證該單例模式是否線程安全...") # 執行結果 開始驗證該單例模式生成的實例ID是否唯一... Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> Created Object: <__main__.Singleton object at 0x01EE5D50> 結束驗證該單例模式生成的實例ID是否唯一... ================================================================ 開始驗證該單例模式是否線程安全... 期望的zero值: 0, 實際的zero值: -9 該實例的實現是否是線程安全的: False 結束驗證該單例模式是否線程安全... >>>
從測試結果來看該單例模式實現只能保證實例唯一,並不能保證線程安全。
線程安全的單例實現
保證單例實現的線程安全需要加鎖進行同步,python提供了兩種鎖threading.Lock() & threading.RLock()實現線程同步,各自有兩種加鎖方式with func.__lock__和lock_inst.acquire() & lock_inst.release().插樁代碼。
threading.Lock() & threading.RLock()
threading.Lock():多線程中基本的鎖對象,同一線程中只能一次acquire,其余的鎖請求只能等鎖釋放後才能獲取。使用過程中可能造成叠代死鎖
threading.RLock():可重入鎖,同一線程中可以多次acquire,acquire()和release()必須成對出現。使用過程中不會造成叠代死鎖
threading.RLock()可以看成是threading.Lock()的改進版。
方式一:裝飾器@synchronized實現同步操作
def synchronized(func): func.__lock__ = threading.RLock() def lock_func(*args, **kwargs): with func.__lock__: return func(*args, **kwargs) return lock_func
在需要同步的方法前添加@synchronized,以上面的單例實現為例。修改後線程安全的單例模式實現,同樣保存在mysingleton.py中:
def synchronized(func): func.__lock__ = threading.RLock() def lock_func(*args, **kwargs): with func.__lock__: return func(*args, **kwargs) return lock_func class Singleton(object): def __init__(self): self.zero = 0 @synchronized def __new__(cls, *args, **kw): if not hasattr(cls, "_instance"): cls._instance = super(Singleton, cls).__new__(cls, *args, **kw) return cls._instance @synchronized def change_zero(self): for i in range(10000000): self.zero += 1 self.zero -= 1
同一個封裝校驗類檢驗是否線程安全,並多次執行避免偶然性。
from mysingleton import Singleton import threading class SingletonCheck(object): def __init__(self): self.task = Singleton() def action(self): obj = Singleton() print ("Created Object: {}".format(obj)) def only_instance_test(self): for i in range(10): t = threading.Thread(target=self.action) t.start() t.join() def thread_safety_test(self): t1 = threading.Thread(target=self.task.change_zero) t2 = threading.Thread(target=self.task.change_zero) t3 = threading.Thread(target=self.task.change_zero) t1.start() t2.start() t3.start() t1.join() t2.join() t3.join() print ("期望的zero值: 0, 實際的zero值: {}\n該實例的實現是否是線程安全的: {}".format(self.task.zero, 0 == self.task.zero)) if __name__ == "__main__": singleton_check = SingletonCheck() print ("開始驗證該單例模式生成的實例ID是否唯一...") singleton_check.only_instance_test() print ("結束驗證該單例模式生成的實例ID是否唯一...") print ("=" * 64) print ("開始驗證該單例模式是否線程安全...") singleton_check.thread_safety_test() print ("結束驗證該單例模式是否線程安全...") # 執行結果 開始驗證該單例模式生成的實例ID是否唯一... Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> Created Object: <__main__.Singleton object at 0x01E65D10> 結束驗證該單例模式生成的實例ID是否唯一... ================================================================ 開始驗證該單例模式是否線程安全... 期望的zero值: 0, 實際的zero值: 0 該實例的實現是否是線程安全的: True 結束驗證該單例模式是否線程安全...
方式二:需同步代碼的前後插入lock_inst.acquire() & lock_inst.release()
class Singleton(object): __rlock = threading.RLock() def __init__(self): self.zero = 0 def __new__(cls, *args, **kw): Singleton.__rlock.acquire() if not hasattr(cls, "_instance"): cls._instance = super(Singleton, cls).__new__(cls, *args, **kw) Singleton.__rlock.release() return cls._instance def change_zero(self): for i in range(10000000): Singleton.__rlock.acquire() self.zero += 1 self.zero -= 1 Singleton.__rlock.release()
校驗過程可參考方式一。
註意事項
- 單例應該只用來保存全局狀態,不應該和任何小於程序完整生命周期的作用域綁定
- 不能用反射、序列化、克隆等破壞單例唯一性的方式創建單例,否則只會實例化另一個新對象
- 單例對象一旦創建就會永久駐留內存直到程序終止,過多的單例會增大內存消耗
- 多線程環境下使用單例模式一定要保證線程安全
單例模式的實現
python本身是一門非常靈活的語言,內置許多特性,所以有很多方式實現單例模式,或者達到實例唯一的效果。如下重點介紹python內置的哪些特性可以實現單例模式,所以選擇單線程下的實現較為簡便,如果需要相應修改為多線程下安全的單例模式,請參考【單例模式的校驗】中【線程安全】部分給需要同步的代碼加鎖。
1. 重寫__new__()
原理分析
__new__()方法在python中屬於內置函數用於創建類實例,通過重載__new__()方法可以在創建實例過程中自定義我們需要的功能。__new__()和__init__()是剛接觸python是比較容易混淆的兩個方法,二者的區別是__new__()用於創建實例,__init__()用於實例創建後的初始化工作,這一點具備編程基礎的人應該好理解。
__new__()方法至少要有一個參數cls, 參數cls表示當前正在實例化的類。要想得到當前類的實例,應當在當前類的__new__()方法中調用父類的__new__()方法(即super(Singleton, cls).__new__(cls, *args, **kw)),如果父類是object的話,前面括號裏的內容可以直接寫成object.__new__(cls, *args, **kw)
使用__new__()方法創建單例過程:先判斷當前類是否已存在類變量_instance,如果存在直接return該類變量,如果不存在則生成一個Singleton實例,再賦值給_instance,然後return該類變量。
單線程實現
class Singleton(object): def __new__(cls, *args, **kw): if not hasattr(cls, "_instance"): cls._instance = super(Singleton, cls).__new__(cls, *args, **kw) return cls._instance
2. 重寫__call__()
原理分析
在python中,函數其實也是對象,所有的函數都是可調用對象。python中有個內置方法__call__(),如果實現了它,一個類實例也可以變成可調用對象,相當於重載了()運算符。 __call__()的作用是使實例能夠像函數一樣被調用,這是一件比較有意思的事,通過Singleton = Singleton()就可以實現自循環的單例調用,同時不影響實例本身的生命周期,如__new__()和__init__()的過程。
單線程實現
class Singleton(object): def test(self): print(‘Singleton Test‘) def __call__(self, *args, **kwargs): return self Singleton = Singleton()
3. @staticmethod特性
原理分析
單例模式的實質就是提供實例化的唯一通道,實例化的途徑是首先通過自身類的__new__來實例化及__init__進行初始化,如果自身類的實例化被禁止,則可以層層上溯同父類.__new__()及__init__()進行實例化。
在自身類的__init__()中拋出語法錯誤禁止實例化,同時在@staticmethod修改的靜態方法中通過調用當前類的父類.__new__()生成實例,然後把該實例賦值給當前類的類變量_instance,即可以實現實例化的唯一通道。
這種方式可以實現單例模式,但是強制__init__()拋出語法錯誤的做法比較強制,有點粗暴幹涉的意味,不太推薦此種方式,但是知道這種原理就可以了。
單線程實現
class Singleton(object): _instance = None def __init__(self): raise SyntaxError(‘instantiation error, please use get_instance()‘) @staticmethod def get_instance(): if Singleton._instance is None: Singleton._instance = object.__new__(Singleton) return Singleton._instance
4. @classmethod特性
原理分析
使用@classmethod特性實現單例模式的原理跟使用@staticmethod實現單例模式的原理頗為相似,這裏主要介紹下@staticmethod和@classmethod的區別。
- 傳遞參數: @classmethod修飾的方法必須使用類對象作為第一個參數,@staticmethod修飾的方法則不需要傳遞任何參數
- 調用方式: 二者修飾的方法都可以通過類名/實例對象來調用
- 調用對象:
- @classmethod修飾的方法持有cls參數,可以通過cls.xxx的方式調用類的變量、方法、實例等屬性,避免硬編碼;
- @staticmethod修飾的方法中如果想調用類的屬性(變量,方法,實例等)只能通過類名.屬性名(顯式硬編碼)的方式調用
- 使用場景:
- @classmethod: 用在需要訪問當前類屬性的方法,常作為當前類構造函數的補充
- @staticmethod: 確認當前類的某個方法不會涉及到與當前類屬性有關的操作時,可以考慮將該方法定義為當前類的staticmethod
- 子類繼承:
- 有子類繼承時,調用@classmethod修飾的方法中自動傳入的類變量cls是子類,而非父類,
- 有子類繼承時,如果@staticmethod修飾的方法中包含顯式的類名引用,則子類中不會自動替換為其子類名
單線程實現
class Singleton(object):
_instance = None
def __init__(self):
raise SyntaxError(‘instantiation error, please use get_instance()‘)
@classmethod
def get_instance(cls):
if Singleton._instance is None:
Singleton._instance = object.__new__(Singleton)
return Singleton._instance
5. 類屬性
原理分析
python中的屬性是一個比較寬泛的概念,比java中的屬性範圍大得多,也不同於python內置的@property,雖然property翻譯過來也是屬性的意思,但是python的屬性和@property完全不是一回事,類中的變量、方法、實例,實例中的變量都可以成為python的屬性。python屬於動態語言,很多特性與靜態語言不同,所以不能把靜態語言的概念和規則套到python上來。
python中對於屬性的調用,通常采用類.屬性或者實例.屬性的形式,舉個小例子可能更容易理解python的屬性。
class Test(object): x = 10 def foo(): return Test.x >>> Test.x 10 >>> Test.foo() 10
通過Test.x及Test.foo()的方式可以類屬性的方式調用,此處單例模式的實現也是一樣,雖然使用類屬性這個特性實現的單例和通過@staticmethod / @classmethod特性實現的單例從代碼角度來說很相似,但是調用原理完全不同。
單線程實現
class Singleton(object): _instance = None def __init__(self): raise SyntaxError(‘can not instance, please use get_instance‘) def get_instance(): if Singleton._instance is None: Singleton._instance = object.__new__(Singleton) return Singleton._instance
6. 元類__metaclass__
原理分析
python中的metaclass比較復雜晦澀,不打算大篇幅介紹,這裏主要應用元類的兩點特性:
- type(name, bases, dict): 動態創建類。
- name: 類的名稱
- bases: 基類的元組
- dict: 類內定義的namespace變量
- __metaclass__: 自定義元類
我們可以通過type定義一個元類,並把它返回的類(新建的元類)賦值給另一個類的__metaclass__屬性,這樣另一個類就具備了type創建的類的所有特性。
單線程實現
class Singleton(type): def __init__(cls, name, bases, dict): super(Singleton, cls).__init__(name, bases, dict) cls._instance = None def __call__(cls, *args, **kw): if cls._instance is None: cls._instance = super(Singleton, cls).__call__(*args, **kw) return cls._instance
class MySingletonClass(object):
__metaclass__ = Singleton
# 或者使用如下格式 # class MySingletonClass(metaclass=Singleton): # pass
7. 方法裝飾器
原理分析
裝飾器是基於AOP實現的,可以動態地改變類或函數的功能,具有很高的解耦性和靈活性,在python中應用非常廣泛。
用函數裝飾器實現單例模式的原理是:先創建一個可以傳入類對象的外層函數,在該外層函數中創建一個instance字典來保存單例,同時創建一個內層函數get_instance(用來返回單例),在該內層函數判斷instance字典中是否包含單例,如果沒有就創建單例並以鍵值對的形式保存在instance字典中,然後通過get_instance()返回單例,外層函數的返回值為內層函數名。
使用函數裝飾器時記住添加functools模塊中的@wraps修復函數名及文檔屬性。
單線程實現
from functools import wraps def singleton(cls): _instance = {} @wraps(cls) def get_instance(*args, **kw): if cls not in _instance: _instance[cls] = cls(*args, **kw) return _instance[cls] return get_instance @singleton class MySingletonClass(object): pass
8. 類裝飾器
原理分析
裝飾器本身就是對所修飾的類、對象、函數功能的擴展,相當於按需定制、重新封裝。如果想讓裝飾器正常工作,必須在裝飾器內部返回一個可調用的對象,可調用的對象可以是函數,也可以是類,所以裝飾器不僅可以是函數,也可以是類。
類裝飾器主要是通過類的構造函數__init__()傳入一個函數或類對象,然後重載__call__()並且返回一個函數或類對象,來達到裝飾器的目的。
此處使用類裝飾器實現單例模式通過__init__()傳入的是類對象,重載__call__()返回的也是類對象。將自定義的@Singleton裝飾器附加到MySingletonClass類上就會調用單例模式生成MySingletonClass的唯一實例。
單線程實現
class Singleton(object): def __init__(self, cls): self._cls = cls self._instance = {} def __call__(self, *args, **kwargs): if self._cls not in self._instance: self._instance[self._cls] = self._cls(*args, **kwargs) return self._instance[self._cls] @Singleton class MySingletonClass(object): pass
9. import 模塊
原理分析
分析python模塊加載機制之前先回顧一下java加載機制對於static的處理,我們知道靜態代碼塊、靜態屬性和靜態方法只會在類首次加載的時候初始化一次,而且是全局性的、不會加載第二次,後續直接調用即可。
python模塊的加載也非常類似,第一次import module時系統會執行模塊代碼生成.pyc文件,第二次import時就會直接加載.pyc文件,不會再次執行模塊代碼。所以,如果我們把需要單例化的函數和數據封裝在一個模塊文件中import,就可以獲得這個單例對象了。從import module的機制來說,python的模塊天然支持單例模式。
由import module方式生成的單例需要在代碼中先指定單例,可作為簡單使用,靈活性較差。
單線程實現
將如下代碼保存為mysingleton.py,並執行from mysingleton import singleton導入單例。
class Singleton(object):
def test(self):
print ("for test")
singleton = Singleton()
10. 共享屬性
原理分析
本文開始就提到單例模式一個比較常用的場景就是提供全局訪問點,可以存放與程序生命周期一致的全局共享資源,所以有時候我們並不在意是否真正實例唯一,更關心的是能否提供所有實例共享的狀態或資源。
在python中每個對象都有自己的命名空間,空間內的變量存儲在對象的__dict__中。類本身也是對象,所以類也有自己的__dict__,類的__dict__是由類的所有實例共享的,對於類中任一實例的屬性的修改,所有的實例都會受到影響。
根據類的__dict__特性生成的實例並不唯一,實例id也會不同,嚴格地說並不屬於單例模式的範疇,但此處更看重的是與單例模式一致的提供訪問全局共享資源的功能。
單線程實現
class Borg(object): _shared_state = {} def __init__(self): self.__dict__ = self._shared_state class MySingletonClass(Borg): def __init__(self, name): super(MySingletonClass, self).__init__() self.name = name def __str__(self): return self.name >>> a = MySingletonClass("first") >>> b = MySingletonClass("second") >>> print (a, b) second second >>> print (id(a), id(b)) 34292112 34197072 >>>
Python設計模式 - 創建型 - 單例模式(Singleton) - 十種